feat: validate catalog

GreptimeTeam · May 21, 2024 · e48d097 · e48d097
1 parent 955b3e3
commit e48d097
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 21 deletions.
diff --git a/src/common/function/src/table/flush_compact_table.rs b/src/common/function/src/table/flush_compact_table.rs
@@ -203,6 +203,7 @@ mod tests {
     use std::sync::Arc;
 
     use api::v1::region::compact_request::Options;
+    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use common_query::prelude::TypeSignature;
     use datatypes::vectors::{StringVector, UInt64Vector};
     use session::context::QueryContext;
@@ -288,53 +289,56 @@ mod tests {
             (
                 &["table"],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "public".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::Regular(Default::default()),
                 },
             ),
             (
-                &["a.table"],
+                &[&format!("{}.table", DEFAULT_SCHEMA_NAME)],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "a".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::Regular(Default::default()),
                 },
             ),
             (
-                &["a.b.c"],
+                &[&format!(
+                    "{}.{}.table",
+                    DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME
+                )],
                 CompactTableRequest {
-                    catalog_name: "a".to_string(),
-                    schema_name: "b".to_string(),
-                    table_name: "c".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
+                    table_name: "table".to_string(),
                     compact_options: Options::Regular(Default::default()),
                 },
             ),
             (
                 &["table", "regular"],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "public".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::Regular(Default::default()),
                 },
             ),
             (
                 &["table", "strict_window"],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "public".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::StrictWindow(StrictWindow { window_seconds: 0 }),
                 },
             ),
             (
                 &["table", "strict_window", "3600"],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "public".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::StrictWindow(StrictWindow {
                         window_seconds: 3600,
@@ -344,8 +348,8 @@ mod tests {
             (
                 &["table", "regular", "abcd"],
                 CompactTableRequest {
-                    catalog_name: "greptime".to_string(),
-                    schema_name: "public".to_string(),
+                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+                    schema_name: DEFAULT_SCHEMA_NAME.to_string(),
                     table_name: "table".to_string(),
                     compact_options: Options::Regular(Default::default()),
                 },
@@ -360,5 +364,14 @@ mod tests {
             &QueryContext::arc(),
         )
         .is_err());
+
+        assert!(parse_compact_params(
+            &["a.b.table", "strict_window", "abc"]
+                .into_iter()
+                .map(ValueRef::String)
+                .collect::<Vec<_>>(),
+            &QueryContext::arc(),
+        )
+        .is_err());
     }
 }
diff --git a/src/session/src/table_name.rs b/src/session/src/table_name.rs
@@ -12,20 +12,30 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use snafu::ensure;
 use sql::ast::ObjectName;
-use sql::error::{InvalidSqlSnafu, Result};
+use sql::error::{InvalidSqlSnafu, PermissionDeniedSnafu, Result};
 use sql::parser::ParserContext;
 
 use crate::QueryContextRef;
 
-/// Parse table name into `(catalog, schema, table)` with query context.
+/// Parse table name into `(catalog, schema, table)` with query context and validates
+/// if catalog matches current catalog in query context.
 pub fn table_name_to_full_name(
     name: &str,
     query_ctx: &QueryContextRef,
 ) -> Result<(String, String, String)> {
     let obj_name = ParserContext::parse_table_name(name, query_ctx.sql_dialect())?;
 
-    table_idents_to_full_name(&obj_name, query_ctx)
+    let (catalog, schema, table) = table_idents_to_full_name(&obj_name, query_ctx)?;
+    ensure!(
+        catalog == query_ctx.current_catalog(),
+        PermissionDeniedSnafu {
+            target: catalog,
+            current: query_ctx.current_catalog(),
+        }
+    );
+    Ok((catalog, schema, table))
 }
 
 /// Converts maybe fully-qualified table name (`<catalog>.<schema>.<table>`) to tuple.

diff --git a/src/sql/src/error.rs b/src/sql/src/error.rs
@@ -213,6 +213,18 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display(
+        "Permission denied while operating catalog {} from current catalog {}",
+        target,
+        current
+    ))]
+    PermissionDenied {
+        target: String,
+        current: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -241,7 +253,8 @@ impl ErrorExt for Error {
             | InvalidSqlValue { .. }
             | TimestampOverflow { .. }
             | InvalidTableOption { .. }
-            | InvalidCast { .. } => StatusCode::InvalidArguments,
+            | InvalidCast { .. }
+            | PermissionDenied { .. } => StatusCode::InvalidArguments,
 
             SerializeColumnDefaultConstraint { source, .. } => source.status_code(),
             ConvertToGrpcDataType { source, .. } => source.status_code(),