refactor: extract storages into sub crates

Cargo.toml

@@ -38,6 +38,8 @@ members = [
     "common/tracing",
     "common/users",
     "common/storage",
+    "common/storages/hive",
+    "common/storages/index",
     "common/fuse-meta",
     # Query
     "query",

common/base/Cargo.toml

@@ -46,4 +46,3 @@ uuid = { version = "1.1.2", features = ["serde", "v4"] }
 
 [dev-dependencies]
 anyhow = "1.0.58"
-common-macros = { path = "../macros" }

common/catalog/Cargo.toml

@@ -20,7 +20,7 @@ common-fuse-meta = { path = "../fuse-meta" }
 common-io = { path = "../io" }
 common-meta-app = { path = "../meta/app" }
 common-meta-types = { path = "../meta/types" }
-common-pipeline = { path = "../pipeline" }
+common-pipeline-core = { path = "../pipeline/core" }
 common-planners = { path = "../planners" }
 common-settings = { path = "../settings" }
 common-users = { path = "../users" }

common/catalog/src/catalog.rs

@@ -49,6 +49,8 @@ use crate::table::Table;
 use crate::table_args::TableArgs;
 use crate::table_function::TableFunction;
 
+pub const CATALOG_DEFAULT: &str = "default";
+
 pub struct CatalogManager {
     pub catalogs: HashMap<String, Arc<dyn Catalog>>,
 }

common/catalog/src/table.rs

@@ -25,7 +25,7 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_app::schema::TableInfo;
 use common_meta_types::MetaId;
-use common_pipeline::Pipeline;
+use common_pipeline_core::Pipeline;
 use common_planners::DeletePlan;
 use common_planners::Expression;
 use common_planners::Extras;

common/legacy-parser/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "common-legacy-parser"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[lib]
+doctest = false
+test = false
+
+[dependencies]
+common-ast = { path = "../ast" }
+common-datavalues = { path = "../datavalues" }
+common-exception = { path = "../exception" }
+common-functions = { path = "../functions" }
+common-meta-types = { path = "../meta/types" }
+common-planners = { path = "../planners" }
+common-storage = { path = "../storage" }
+
+async-trait = "0.1.56"
+clap = { version = "3.2.5", features = ["derive", "env"] }
+futures = "0.3.21"
+futures-util = "0.3.21"
+itertools = "0.10.3"
+metrics = "0.19.0"
+num_cpus = "1.13.1"
+once_cell = "1.12.0"
+opendal = { version = "0.11.4", features = ["retry"] }
+parking_lot = "0.12.1"
+serde = { version = "1.0.137", features = ["derive"] }
+sqlparser = { git = "https://github.com/datafuse-extras/sqlparser-rs", rev = "7f246e3" }
+time = "0.3.10"

query/src/sql/statements/analyzer_expr_sync.rs → common/legacy-parser/src/analyzer/analyzer_expr_sync.rs

@@ -35,9 +35,9 @@ use sqlparser::ast::Query;
 use sqlparser::ast::UnaryOperator;
 use sqlparser::ast::Value;
 
-use crate::sessions::SessionType;
-use crate::sql::statements::analyzer_value_expr::ValueExprAnalyzer;
-use crate::sql::SQLCommon;
+use crate::analyzer_value_expr::ValueExprAnalyzer;
+use crate::sql_common::SQLCommon;
+use crate::sql_dialect::SQLDialect;
 
 #[derive(Clone)]
 pub struct ExpressionSyncAnalyzer {}
@@ -53,7 +53,7 @@ impl ExpressionSyncAnalyzer {
         // Build RPN for expr. Because async function unsupported recursion
         for rpn_item in &ExprRPNBuilder::build(expr, vec![])? {
             match rpn_item {
-                ExprRPNItem::Value(v) => Self::analyze_value(v, &mut stack, SessionType::MySQL)?,
+                ExprRPNItem::Value(v) => Self::analyze_value(v, &mut stack, SQLDialect::MySQL)?,
                 ExprRPNItem::Identifier(v) => self.analyze_identifier(v, &mut stack)?,
                 ExprRPNItem::QualifiedIdentifier(v) => self.analyze_identifiers(v, &mut stack)?,
                 ExprRPNItem::Function(v) => self.analyze_function(v, &mut stack)?,
@@ -91,7 +91,11 @@ impl ExpressionSyncAnalyzer {
         }
     }
 
-    fn analyze_value(value: &Value, args: &mut Vec<Expression>, typ: SessionType) -> Result<()> {
+    fn analyze_value(
+        value: &Value,
+        args: &mut Vec<Expression>,
+        typ: impl Into<SQLDialect>,
+    ) -> Result<()> {
         args.push(ValueExprAnalyzer::analyze(value, typ)?);
         Ok(())
     }

query/src/sql/statements/analyzer_value_expr.rs → common/legacy-parser/src/analyzer/analyzer_value_expr.rs

@@ -19,12 +19,12 @@ use common_planners::Expression;
 use sqlparser::ast::DateTimeField;
 use sqlparser::ast::Value;
 
-use crate::sessions::SessionType;
+use crate::sql_dialect::SQLDialect;
 
 pub struct ValueExprAnalyzer;
 
 impl ValueExprAnalyzer {
-    pub fn analyze(value: &Value, typ: SessionType) -> Result<Expression> {
+    pub fn analyze(value: &Value, typ: impl Into<SQLDialect>) -> Result<Expression> {
         match value {
             Value::Null => Self::analyze_null_value(),
             Value::Boolean(value) => Self::analyze_bool_value(value),
@@ -35,7 +35,7 @@ impl ValueExprAnalyzer {
                 // Only MySQL dialect Support insert SQL like this:
                 // INSERT INTO t VALUES("val");
                 // https://github.com/datafuselabs/databend/issues/4861
-                if let SessionType::MySQL = typ {
+                if let SQLDialect::MySQL = typ.into() {
                     Self::analyze_string_value(value)
                 } else {
                     Result::Err(ErrorCode::SyntaxException(format!(

query/src/storages/hive/mod.rs → common/legacy-parser/src/analyzer/mod.rs

@@ -1,4 +1,4 @@
-//  Copyright 2021 Datafuse Labs.
+//  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.
@@ -12,5 +12,5 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-pub mod hive_parquet_block_reader;
-pub use hive_parquet_block_reader::HiveParquetBlockReader;
+pub mod analyzer_expr_sync;
+pub mod analyzer_value_expr;

query/src/pipelines/processors/sinks/mod.rs → common/legacy-parser/src/lib.rs

@@ -12,18 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod async_sink;
-mod context_sink;
-mod empty_sink;
-mod subquery_receive_sink;
-mod sync_sink;
-mod sync_sink_sender;
+pub use analyzer::analyzer_expr_sync;
+pub use analyzer::analyzer_value_expr;
+mod analyzer;
+mod parser;
+pub mod sql_common;
+mod sql_dialect;
 
-pub use async_sink::AsyncSink;
-pub use async_sink::AsyncSinker;
-pub use context_sink::ContextSink;
-pub use empty_sink::EmptySink;
-pub use subquery_receive_sink::SubqueryReceiveSink;
-pub use sync_sink::Sink;
-pub use sync_sink::Sinker;
-pub use sync_sink_sender::SyncSenderSink;
+pub use parser::ExprParser;
+pub use parser::ExpressionParser;
+pub use sql_dialect::SQLDialect;

common/legacy-parser/src/parser/expr_parser.rs

@@ -0,0 +1,45 @@
+//  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 sqlparser::ast::Expr;
+use sqlparser::dialect::MySqlDialect;
+use sqlparser::parser::Parser;
+use sqlparser::parser::ParserError;
+use sqlparser::tokenizer::Token;
+use sqlparser::tokenizer::Tokenizer;
+
+pub struct ExprParser;
+
+impl ExprParser {
+    pub fn parse_expr(expr: &str) -> Result<Expr, ParserError> {
+        let dialect = &MySqlDialect {};
+        let mut tokenizer = Tokenizer::new(dialect, expr);
+        let (tokens, position_map) = tokenizer.tokenize()?;
+        let mut parser = Parser::new(tokens, position_map, dialect);
+        parser.parse_expr()
+    }
+
+    pub fn parse_exprs(expr: &str) -> Result<Vec<Expr>, ParserError> {
+        let dialect = &MySqlDialect {};
+        let mut tokenizer = Tokenizer::new(dialect, expr);
+        let (tokens, position_map) = tokenizer.tokenize()?;
+        let mut parser = Parser::new(tokens, position_map, dialect);
+
+        parser.expect_token(&Token::LParen)?;
+        let exprs = parser.parse_comma_separated(Parser::parse_expr)?;
+        parser.expect_token(&Token::RParen)?;
+
+        Ok(exprs)
+    }
+}

common/legacy-parser/src/parser/expression_parser.rs

@@ -0,0 +1,40 @@
+//  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 common_exception::Result;
+use common_planners::Expression;
+
+use crate::analyzer_expr_sync::ExpressionSyncAnalyzer;
+use crate::ExprParser;
+
+pub struct ExpressionParser;
+
+impl ExpressionParser {
+    pub fn parse_expr(expr: &str) -> Result<Expression> {
+        let expr = ExprParser::parse_expr(expr)?;
+        let analyzer = ExpressionSyncAnalyzer::create();
+        analyzer.analyze(&expr)
+    }
+
+    pub fn parse_exprs(expr: &str) -> Result<Vec<Expression>> {
+        let exprs = ExprParser::parse_exprs(expr)?;
+        let analyzer = ExpressionSyncAnalyzer::create();
+
+        let results = exprs
+            .iter()
+            .map(|expr| analyzer.analyze(expr))
+            .collect::<Result<Vec<_>>>();
+        results
+    }
+}

query/src/storages/fuse/mod.rs → common/legacy-parser/src/parser/mod.rs

@@ -1,4 +1,4 @@
-//  Copyright 2021 Datafuse Labs.
+//  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.
@@ -12,15 +12,8 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-mod constants;
-mod fuse_part;
-mod fuse_table;
-pub mod io;
-pub mod operations;
-pub mod pruning;
-pub mod statistics;
-pub mod table_functions;
+mod expr_parser;
+mod expression_parser;
 
-pub use constants::*;
-pub use fuse_part::ColumnLeaf;
-pub use fuse_table::FuseTable;
+pub use expr_parser::ExprParser;
+pub use expression_parser::ExpressionParser;

common/legacy-parser/src/sql_dialect.rs

@@ -0,0 +1,18 @@
+//  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.
+
+pub enum SQLDialect {
+    MySQL,
+    Other,
+}

common/pipeline/Cargo.toml → common/pipeline/core/Cargo.toml

@@ -1,6 +1,9 @@
 [package]
-name = "common-pipeline"
+name = "common-pipeline-core"
 version = "0.1.0"
+authors = ["Databend Authors <opensource@datafuselabs.com>"]
+license = "Apache-2.0"
+publish = false
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -9,10 +12,10 @@ doctest = false
 test = false
 
 [dependencies]
-common-datablocks = { path = "../datablocks" }
-common-datavalues = { path = "../datavalues" }
-common-exception = { path = "../exception" }
-common-io = { path = "../io" }
+common-datablocks = { path = "../../datablocks" }
+common-datavalues = { path = "../../datavalues" }
+common-exception = { path = "../../exception" }
+common-io = { path = "../../io" }
 
 async-trait = "0.1.56"
 futures = "0.3.21"
@@ -24,3 +27,6 @@ parking_lot = "0.12.1"
 petgraph = "0.6.2"
 serde = { version = "1.0.137", features = ["derive"] }
 time = "0.3.10"
+
+[dev-dependencies]
+tokio = { version = "1.19.2", features = ["full"] }

common/pipeline/core/tests/it/main.rs

@@ -0,0 +1,14 @@
+// Copyright 2021 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.
+mod pipelines;

common/pipeline/core/tests/it/pipelines/mod.rs

@@ -0,0 +1,15 @@
+// Copyright 2021 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.
+
+mod processors;

common/pipeline/core/tests/it/pipelines/processors/mod.rs

@@ -0,0 +1,15 @@
+// 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.
+
+mod port_test;