Merge pull request #6975 from lichuang/share_sql

feat: add Create Share sql support

Author: mergify[bot]

common/ast/src/ast/statements/mod.rs

@@ -19,6 +19,7 @@ mod explain;
 mod insert;
 mod kill;
 mod presign;
+mod share;
 mod show;
 mod stage;
 mod statement;
@@ -33,6 +34,7 @@ pub use explain::*;
 pub use insert::*;
 pub use kill::*;
 pub use presign::*;
+pub use share::*;
 pub use show::*;
 pub use stage::*;
 pub use statement::*;

common/ast/src/ast/statements/share.rs

@@ -0,0 +1,39 @@
+// 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::fmt::Display;
+use std::fmt::Formatter;
+
+use crate::ast::Identifier;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreateShareStmt<'a> {
+    pub if_not_exists: bool,
+    pub share: Identifier<'a>,
+    pub comment: Option<String>,
+}
+
+impl Display for CreateShareStmt<'_> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "CREATE SHARE ")?;
+        if self.if_not_exists {
+            write!(f, "IF NOT EXISTS ")?;
+        }
+        write!(f, "{}", self.share)?;
+        if let Some(comment) = &self.comment {
+            write!(f, " COMMENT = '{comment}'")?;
+        }
+        Ok(())
+    }
+}

common/ast/src/ast/statements/statement.rs

@@ -157,6 +157,9 @@ pub enum Statement<'a> {
     },
 
     Presign(PresignStmt),
+
+    // share
+    CreateShare(CreateShareStmt<'a>),
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -358,6 +361,7 @@ impl<'a> Display for Statement<'a> {
             Statement::DescribeStage { stage_name } => write!(f, "DESC STAGE {stage_name}")?,
             Statement::Call(stmt) => write!(f, "{stmt}")?,
             Statement::Presign(stmt) => write!(f, "{stmt}")?,
+            Statement::CreateShare(stmt) => write!(f, "{stmt}")?,
         }
         Ok(())
     }

common/ast/src/parser/statement.rs

@@ -741,6 +741,23 @@ pub fn statement(i: Input) -> IResult<StatementMsg> {
         },
     );
 
+    // share statements
+    let create_share = map(
+        rule! {
+            CREATE ~ SHARE ~ (IF ~ NOT ~ EXISTS )? ~ #ident ~ ( COMMENT ~ "=" ~ #literal_string)?
+        },
+        |(_, _, opt_if_not_exists, share, comment_opt)| {
+            Statement::CreateShare(CreateShareStmt {
+                if_not_exists: opt_if_not_exists.is_some(),
+                share,
+                comment: match comment_opt {
+                    Some(opt) => Some(opt.2),
+                    None => None,
+                },
+            })
+        },
+    );
+
     let statement_body = alt((
         rule!(
             #map(query, |query| Statement::Query(Box::new(query)))
@@ -826,6 +843,10 @@ pub fn statement(i: Input) -> IResult<StatementMsg> {
         rule!(
             #presign: "`PRESIGN [{DOWNLOAD | UPLOAD}] <location> [EXPIRE = 3600]`"
         ),
+        // share
+        rule!(
+            #create_share: "`CREATE SHARE <share_name> [ COMMENT = '<string_literal>' ]`"
+        ),
     ));
 
     map(

common/ast/src/parser/token.rs

@@ -607,6 +607,8 @@ pub enum TokenKind {
     UPDATE,
     #[token("UPLOAD", ignore(ascii_case))]
     UPLOAD,
+    #[token("SHARE", ignore(ascii_case))]
+    SHARE,
     #[token("SUPER", ignore(ascii_case))]
     SUPER,
     #[token("STATUS", ignore(ascii_case))]

common/ast/tests/it/parser.rs

@@ -252,6 +252,8 @@ fn test_statement() {
         r#"PRESIGN @my_stage/path/to/file"#,
         r#"PRESIGN DOWNLOAD @my_stage/path/to/file"#,
         r#"PRESIGN UPLOAD @my_stage/path/to/file EXPIRE=7200"#,
+        r#"CREATE SHARE t COMMENT='share comment';"#,
+        r#"CREATE SHARE IF NOT EXISTS t;"#,
     ];
 
     for case in cases {

common/ast/tests/it/testdata/statement.txt

@@ -5802,3 +5802,41 @@ Presign(
 )
 
 
+---------- Input ----------
+CREATE SHARE t COMMENT='share comment';
+---------- Output ---------
+CREATE SHARE t COMMENT = 'share comment'
+---------- AST ------------
+CreateShare(
+    CreateShareStmt {
+        if_not_exists: false,
+        share: Identifier {
+            name: "t",
+            quote: None,
+            span: Ident(13..14),
+        },
+        comment: Some(
+            "share comment",
+        ),
+    },
+)
+
+
+---------- Input ----------
+CREATE SHARE IF NOT EXISTS t;
+---------- Output ---------
+CREATE SHARE IF NOT EXISTS t
+---------- AST ------------
+CreateShare(
+    CreateShareStmt {
+        if_not_exists: true,
+        share: Identifier {
+            name: "t",
+            quote: None,
+            span: Ident(27..28),
+        },
+        comment: None,
+    },
+)
+
+

common/users/src/user_api.rs

@@ -29,16 +29,19 @@ use common_management::UdfMgr;
 use common_management::UserApi;
 use common_management::UserMgr;
 use common_meta_api::KVApi;
+use common_meta_store::MetaStore;
 use common_meta_store::MetaStoreProvider;
 
 pub struct UserApiProvider {
+    meta: MetaStore,
     client: Arc<dyn KVApi>,
 }
 
 impl UserApiProvider {
     pub async fn create_global(conf: RpcClientConf) -> Result<Arc<UserApiProvider>> {
         let client = MetaStoreProvider::new(conf).try_get_meta_store().await?;
         Ok(Arc::new(UserApiProvider {
+            meta: client.clone(),
             client: client.arc(),
         }))
     }
@@ -66,4 +69,8 @@ impl UserApiProvider {
     pub fn get_setting_api_client(&self, tenant: &str) -> Result<Arc<dyn SettingApi>> {
         Ok(Arc::new(SettingMgr::create(self.client.clone(), tenant)?))
     }
+
+    pub fn get_meta_store_client(&self) -> Arc<MetaStore> {
+        Arc::new(self.meta.clone())
+    }
 }

query/src/interpreters/interpreter_factory_v2.rs

@@ -25,6 +25,7 @@ use crate::interpreters::interpreter_copy_v2::CopyInterpreterV2;
 use crate::interpreters::interpreter_presign::PresignInterpreter;
 use crate::interpreters::interpreter_table_create_v2::CreateTableInterpreterV2;
 use crate::interpreters::AlterUserInterpreter;
+use crate::interpreters::CreateShareInterpreter;
 use crate::interpreters::DropUserInterpreter;
 use crate::sessions::QueryContext;
 use crate::sql::plans::Plan;
@@ -250,6 +251,12 @@ impl InterpreterFactoryV2 {
                 *p.clone(),
             )?)),
             Plan::Kill(p) => Ok(Arc::new(KillInterpreter::try_create(ctx, *p.clone())?)),
+
+            // share plans
+            Plan::CreateShare(p) => Ok(Arc::new(CreateShareInterpreter::try_create(
+                ctx,
+                *p.clone(),
+            )?)),
         }
     }
 }

query/src/interpreters/interpreter_share_create.rs

@@ -0,0 +1,55 @@
+// 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::sync::Arc;
+
+use common_exception::Result;
+use common_meta_api::ShareApi;
+use common_streams::DataBlockStream;
+use common_streams::SendableDataBlockStream;
+
+use crate::interpreters::Interpreter;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+use crate::sql::plans::share::CreateSharePlan;
+
+pub struct CreateShareInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: CreateSharePlan,
+}
+
+impl CreateShareInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: CreateSharePlan) -> Result<Self> {
+        Ok(CreateShareInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for CreateShareInterpreter {
+    fn name(&self) -> &str {
+        "CreateShareInterpreter"
+    }
+
+    async fn execute(&self) -> Result<SendableDataBlockStream> {
+        let user_mgr = self.ctx.get_user_manager();
+        let meta_api = user_mgr.get_meta_store_client();
+        meta_api.create_share(self.plan.clone().into()).await?;
+
+        Ok(Box::pin(DataBlockStream::create(
+            self.plan.schema(),
+            None,
+            vec![],
+        )))
+    }
+}

query/src/interpreters/mod.rs

@@ -49,6 +49,7 @@ mod interpreter_role_revoke;
 mod interpreter_select;
 mod interpreter_select_v2;
 mod interpreter_setting;
+mod interpreter_share_create;
 mod interpreter_show_databases;
 mod interpreter_show_functions;
 mod interpreter_show_grants;
@@ -129,6 +130,7 @@ pub use interpreter_role_revoke::RevokeRoleInterpreter;
 pub use interpreter_select::SelectInterpreter;
 pub use interpreter_select_v2::SelectInterpreterV2;
 pub use interpreter_setting::SettingInterpreter;
+pub use interpreter_share_create::CreateShareInterpreter;
 pub use interpreter_show_databases::ShowDatabasesInterpreter;
 pub use interpreter_show_functions::ShowFunctionsInterpreter;
 pub use interpreter_show_grants::ShowGrantsInterpreter;

query/src/sql/planner/binder/ddl/mod.rs

@@ -14,6 +14,7 @@
 
 mod account;
 mod database;
+mod share;
 mod stage;
 mod table;
 mod view;

query/src/sql/planner/binder/ddl/share.rs

@@ -0,0 +1,44 @@
+// 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_ast::ast::*;
+use common_exception::Result;
+
+use crate::sessions::TableContext;
+use crate::sql::binder::Binder;
+use crate::sql::plans::CreateSharePlan;
+use crate::sql::plans::Plan;
+
+impl<'a> Binder {
+    pub(in crate::sql::planner::binder) async fn bind_create_share(
+        &mut self,
+        stmt: &CreateShareStmt<'a>,
+    ) -> Result<Plan> {
+        let CreateShareStmt {
+            if_not_exists,
+            share,
+            comment,
+        } = stmt;
+
+        let share = share.name.to_lowercase();
+
+        let plan = CreateSharePlan {
+            if_not_exists: *if_not_exists,
+            tenant: self.ctx.get_tenant(),
+            share,
+            comment: comment.as_ref().cloned(),
+        };
+        Ok(Plan::CreateShare(Box::new(plan)))
+    }
+}

query/src/sql/planner/binder/mod.rs

@@ -293,6 +293,11 @@ impl<'a> Binder {
                 self.bind_kill_stmt(bind_context, kill_target, object_id.as_str())
                     .await?
             }
+
+            Statement::CreateShare(stmt) => {
+                self.bind_create_share(stmt).await?
+            }
+
         };
         Ok(plan)
     }

query/src/sql/planner/format/display_plan.rs

@@ -93,6 +93,8 @@ impl Plan {
             Plan::SetVariable(p) => Ok(format!("{:?}", p)),
             Plan::UseDatabase(p) => Ok(format!("{:?}", p)),
             Plan::Kill(p) => Ok(format!("{:?}", p)),
+
+            Plan::CreateShare(p) => Ok(format!("{:?}", p)),
         }
     }
 }