Merge branch 'main' into fix-cast-deterministic

Author: mergify[bot]

Makefile

@@ -18,13 +18,14 @@ fmt:
 
 lint:
 	cargo fmt --all
+	cargo clippy --workspace --all-targets -- -D warnings
+
 	# Cargo.toml file formatter(make setup to install)
 	taplo fmt
 	# Python file formatter(make setup to install)
 	yapf -ri tests/
 	# Bash file formatter(make setup to install)
 	shfmt -l -w scripts/*
-	cargo clippy --workspace --all-targets -- -D warnings
 
 lint-yaml:
 	yamllint -f auto .

src/query/service/src/interpreters/access/accessor.rs

@@ -0,0 +1,58 @@
+// 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::collections::HashMap;
+use std::sync::Arc;
+
+use common_exception::Result;
+use common_legacy_planners::PlanNode;
+
+use crate::interpreters::ManagementModeAccess;
+use crate::sessions::QueryContext;
+use crate::sql::plans::Plan;
+
+pub trait AccessChecker: Sync + Send {
+    // Check the access permission for the old plan.
+    // TODO(bohu): Remove after new plan done.
+    fn check(&self, plan: &PlanNode) -> Result<()>;
+
+    // Check the access permission for the old plan.
+    fn check_new(&self, _plan: &Plan) -> Result<()>;
+}
+
+pub struct Accessor {
+    accessors: HashMap<String, Box<dyn AccessChecker>>,
+}
+
+impl Accessor {
+    pub fn create(ctx: Arc<QueryContext>) -> Self {
+        let mut accessors: HashMap<String, Box<dyn AccessChecker>> = Default::default();
+        accessors.insert("management".to_string(), ManagementModeAccess::create(ctx));
+        Accessor { accessors }
+    }
+
+    pub fn check(&self, plan: &PlanNode) -> Result<()> {
+        for accessor in self.accessors.values() {
+            accessor.check(plan)?;
+        }
+        Ok(())
+    }
+
+    pub fn check_new(&self, plan: &Plan) -> Result<()> {
+        for accessor in self.accessors.values() {
+            accessor.check_new(plan)?;
+        }
+        Ok(())
+    }
+}

src/query/service/src/interpreters/access/management_mode_access.rs

@@ -18,6 +18,7 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_legacy_planners::PlanNode;
 
+use crate::interpreters::access::AccessChecker;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
 use crate::sql::plans::Plan;
@@ -27,12 +28,14 @@ pub struct ManagementModeAccess {
 }
 
 impl ManagementModeAccess {
-    pub fn create(ctx: Arc<QueryContext>) -> Self {
-        ManagementModeAccess { ctx }
+    pub fn create(ctx: Arc<QueryContext>) -> Box<dyn AccessChecker> {
+        Box::new(ManagementModeAccess { ctx })
     }
+}
 
+impl AccessChecker for ManagementModeAccess {
     // Check what we can do if in management mode.
-    pub fn check(&self, plan: &PlanNode) -> Result<()> {
+    fn check(&self, plan: &PlanNode) -> Result<()> {
         // Allows for management-mode.
         if self.ctx.get_config().query.management_mode {
             return match plan {
@@ -47,7 +50,7 @@ impl ManagementModeAccess {
     }
 
     // Check what we can do if in management mode.
-    pub fn check_new(&self, plan: &Plan) -> Result<()> {
+    fn check_new(&self, plan: &Plan) -> Result<()> {
         // Allows for management-mode.
         if self.ctx.get_config().query.management_mode {
             let ok = match plan {

src/query/service/src/interpreters/access/mod.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod accessor;
 mod management_mode_access;
 
+pub use accessor::AccessChecker;
+pub use accessor::Accessor;
 pub use management_mode_access::ManagementModeAccess;

src/query/service/src/interpreters/interpreter_factory.rs

@@ -18,6 +18,7 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_legacy_planners::PlanNode;
 
+use crate::interpreters::access::Accessor;
 use crate::interpreters::DeleteInterpreter;
 use crate::interpreters::EmptyInterpreter;
 use crate::interpreters::ExplainInterpreter;
@@ -35,10 +36,14 @@ pub struct InterpreterFactory;
 /// Such as: SelectPlan -> SelectInterpreter, ExplainPlan -> ExplainInterpreter, ...
 impl InterpreterFactory {
     pub fn get(ctx: Arc<QueryContext>, plan: PlanNode) -> Result<Arc<dyn Interpreter>> {
+        // Check the access permission.
+        let access_checker = Accessor::create(ctx.clone());
+        access_checker.check(&plan)?;
+
         let inner = Self::create_interpreter(ctx.clone(), &plan)?;
         let query_kind = plan.name().to_string();
         Ok(Arc::new(InterceptorInterpreter::create(
-            ctx, inner, plan, None, query_kind,
+            ctx, inner, query_kind,
         )))
     }
 

src/query/service/src/interpreters/interpreter_factory_interceptor.rs

@@ -17,13 +17,11 @@ use std::time::SystemTime;
 
 use common_datavalues::DataSchemaRef;
 use common_exception::Result;
-use common_legacy_planners::PlanNode;
 use common_streams::ErrorStream;
 use common_streams::ProgressStream;
 use common_streams::SendableDataBlockStream;
 use parking_lot::Mutex;
 
-use crate::interpreters::access::ManagementModeAccess;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
 use crate::interpreters::InterpreterQueryLog;
@@ -32,34 +30,21 @@ use crate::pipelines::SourcePipeBuilder;
 use crate::sessions::QueryContext;
 use crate::sessions::SessionManager;
 use crate::sessions::TableContext;
-use crate::sql::plans::Plan;
 
 pub struct InterceptorInterpreter {
     ctx: Arc<QueryContext>,
-    plan: PlanNode,
-    new_plan: Option<Plan>,
     inner: InterpreterPtr,
     query_log: InterpreterQueryLog,
     source_pipe_builder: Mutex<Option<SourcePipeBuilder>>,
-    management_mode_access: ManagementModeAccess,
 }
 
 impl InterceptorInterpreter {
-    pub fn create(
-        ctx: Arc<QueryContext>,
-        inner: InterpreterPtr,
-        plan: PlanNode,
-        new_plan: Option<Plan>,
-        query_kind: String,
-    ) -> Self {
+    pub fn create(ctx: Arc<QueryContext>, inner: InterpreterPtr, query_kind: String) -> Self {
         InterceptorInterpreter {
             ctx: ctx.clone(),
-            plan,
-            new_plan,
             inner,
-            query_log: InterpreterQueryLog::create(ctx.clone(), query_kind),
+            query_log: InterpreterQueryLog::create(ctx, query_kind),
             source_pipe_builder: Mutex::new(None),
-            management_mode_access: ManagementModeAccess::create(ctx),
         }
     }
 }
@@ -75,12 +60,6 @@ impl Interpreter for InterceptorInterpreter {
     }
 
     async fn execute(&self, ctx: Arc<QueryContext>) -> Result<SendableDataBlockStream> {
-        // Management mode access check.
-        match &self.new_plan {
-            Some(p) => self.management_mode_access.check_new(p)?,
-            _ => self.management_mode_access.check(&self.plan)?,
-        }
-
         let _ = self
             .inner
             .set_source_pipe_builder((*self.source_pipe_builder.lock()).clone());

src/query/service/src/interpreters/interpreter_factory_v2.rs

@@ -15,12 +15,11 @@
 use std::sync::Arc;
 
 use common_exception::Result;
-use common_legacy_planners::EmptyPlan;
-use common_legacy_planners::PlanNode;
 
 use super::interpreter_share_desc::DescShareInterpreter;
 use super::interpreter_user_stage_drop::DropUserStageInterpreter;
 use super::*;
+use crate::interpreters::access::Accessor;
 use crate::interpreters::interpreter_copy_v2::CopyInterpreterV2;
 use crate::interpreters::interpreter_presign::PresignInterpreter;
 use crate::interpreters::interpreter_table_create_v2::CreateTableInterpreterV2;
@@ -44,13 +43,15 @@ impl InterpreterFactoryV2 {
     }
 
     pub fn get(ctx: Arc<QueryContext>, plan: &Plan) -> Result<InterpreterPtr> {
+        // Check the access permission.
+        let access_checker = Accessor::create(ctx.clone());
+        access_checker.check_new(plan)?;
+
         let inner = InterpreterFactoryV2::create_interpreter(ctx.clone(), plan)?;
 
         Ok(Arc::new(InterceptorInterpreter::create(
             ctx,
             inner,
-            PlanNode::Empty(EmptyPlan::create()),
-            Some(plan.clone()),
             plan.to_string(),
         )))
     }

src/query/service/tests/it/interpreters/access/management_mode_access.rs

@@ -158,14 +158,18 @@ async fn test_management_mode_access() -> Result<()> {
     for group in groups {
         for test in group.tests {
             let (plan, _, _) = planner.plan_sql(test.query).await?;
-            let interpreter = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
-            let res = interpreter.execute(ctx.clone()).await;
-            assert_eq!(
-                test.is_err,
-                res.is_err(),
-                "in test case:{:?}",
-                (group.name, test.name)
-            );
+            if test.is_err {
+                let res = InterpreterFactoryV2::get(ctx.clone(), &plan);
+                assert_eq!(
+                    test.is_err,
+                    res.is_err(),
+                    "in test case:{:?}",
+                    (group.name, test.name)
+                );
+            } else {
+                let interpreter = InterpreterFactoryV2::get(ctx.clone(), &plan)?;
+                interpreter.execute(ctx.clone()).await?;
+            }
         }
     }