Remove depends on DatabasesCatalog

Signed-off-by: Xuanwo <github@xuanwo.io>

Author: Xuanwo

src/query/catalog/src/catalog.rs

@@ -171,4 +171,6 @@ pub trait Catalog: DynClone + Send + Sync {
     fn get_table_engines(&self) -> Vec<StorageDescription> {
         unimplemented!()
     }
+
+    fn is_case_insensitive_db(&self, db: &str) -> bool;
 }

src/query/service/src/catalogs/default/database_catalog.rs

@@ -92,10 +92,6 @@ impl DatabaseCatalog {
         );
         Ok(res)
     }
-
-    pub fn is_case_insensitive_db(db: &str) -> bool {
-        db.to_uppercase() == "INFORMATION_SCHEMA"
-    }
 }
 
 #[async_trait::async_trait]
@@ -104,14 +100,18 @@ impl Catalog for DatabaseCatalog {
         self
     }
 
+    fn is_case_insensitive_db(&self, db: &str) -> bool {
+        db.to_uppercase() == "INFORMATION_SCHEMA"
+    }
+
     async fn get_database(&self, tenant: &str, db_name: &str) -> Result<Arc<dyn Database>> {
         if tenant.is_empty() {
             return Err(ErrorCode::TenantIsEmpty(
                 "Tenant can not empty(while get database)",
             ));
         }
 
-        let db_name = if Self::is_case_insensitive_db(db_name) {
+        let db_name = if self.is_case_insensitive_db(db_name) {
             db_name.to_uppercase()
         } else {
             db_name.to_string()
@@ -243,7 +243,7 @@ impl Catalog for DatabaseCatalog {
             ));
         }
 
-        let (db_name, table_name) = if Self::is_case_insensitive_db(db_name) {
+        let (db_name, table_name) = if self.is_case_insensitive_db(db_name) {
             (db_name.to_uppercase(), table_name.to_uppercase())
         } else {
             (db_name.to_string(), table_name.to_string())
@@ -274,7 +274,7 @@ impl Catalog for DatabaseCatalog {
             ));
         }
 
-        let db_name = if Self::is_case_insensitive_db(db_name) {
+        let db_name = if self.is_case_insensitive_db(db_name) {
             db_name.to_uppercase()
         } else {
             db_name.to_string()
@@ -304,7 +304,7 @@ impl Catalog for DatabaseCatalog {
             ));
         }
 
-        let db_name = if Self::is_case_insensitive_db(db_name) {
+        let db_name = if self.is_case_insensitive_db(db_name) {
             db_name.to_uppercase()
         } else {
             db_name.to_string()

src/query/service/src/catalogs/default/immutable_catalog.rs

@@ -87,6 +87,10 @@ impl Catalog for ImmutableCatalog {
         self
     }
 
+    fn is_case_insensitive_db(&self, _: &str) -> bool {
+        unimplemented!()
+    }
+
     async fn get_database(&self, _tenant: &str, db_name: &str) -> Result<Arc<dyn Database>> {
         match db_name {
             "system" => Ok(self.sys_db.clone()),

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -148,6 +148,10 @@ impl Catalog for MutableCatalog {
         self
     }
 
+    fn is_case_insensitive_db(&self, _: &str) -> bool {
+        unimplemented!()
+    }
+
     async fn get_database(&self, tenant: &str, db_name: &str) -> Result<Arc<dyn Database>> {
         let db_info = self
             .ctx

src/query/service/src/sql/planner/binder/ddl/table.rs

@@ -38,7 +38,6 @@ use common_legacy_planners::*;
 use common_meta_app::schema::TableMeta;
 use tracing::debug;
 
-use crate::catalogs::DatabaseCatalog;
 use crate::sql::binder::scalar::ScalarBinder;
 use crate::sql::binder::Binder;
 use crate::sql::binder::Visibility;
@@ -141,7 +140,11 @@ impl<'a> Binder {
             .map(|ident| normalize_identifier(ident, &self.name_resolution_ctx).name)
             .unwrap_or_else(|| self.ctx.get_current_database());
 
-        if DatabaseCatalog::is_case_insensitive_db(database.as_str()) {
+        if self
+            .ctx
+            .get_catalog(&self.ctx.get_current_catalog())?
+            .is_case_insensitive_db(database.as_str())
+        {
             database = database.to_uppercase();
         }
 

src/query/storages/hive/src/hive_catalog.rs

@@ -196,6 +196,10 @@ impl Catalog for HiveCatalog {
         self
     }
 
+    fn is_case_insensitive_db(&self, _: &str) -> bool {
+        false
+    }
+
     #[tracing::instrument(level = "info", skip(self))]
     async fn get_database(&self, _tenant: &str, db_name: &str) -> Result<Arc<dyn Database>> {
         let client = self.get_client()?;