refactor: Cleanup planner to make space for saying goodbye

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/catalog/src/cluster_info.rs

@@ -20,3 +20,24 @@ pub struct Cluster {
     pub local_id: String,
     pub nodes: Vec<Arc<NodeInfo>>,
 }
+
+impl Cluster {
+    /// If this cluster is empty?
+    ///
+    /// # Note
+    ///
+    /// Cluster empty means:
+    ///
+    /// - There is no active node (is this possible?).
+    /// - There is only one node (myself).
+    ///
+    /// # TODO
+    ///
+    /// From @Xuanwo
+    ///
+    /// Ideally, we should implement a cluster trait to replace `ClusterHelper`
+    /// defined in `databend-query`.
+    pub fn is_empty(&self) -> bool {
+        self.nodes.len() <= 1
+    }
+}

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/interpreters/interpreter_select.rs

@@ -19,7 +19,6 @@ use common_exception::Result;
 use common_legacy_planners::PlanNode;
 use common_legacy_planners::SelectPlan;
 
-use crate::clusters::ClusterHelper;
 use crate::interpreters::plan_schedulers;
 use crate::interpreters::Interpreter;
 use crate::optimizers::Optimizers;

src/query/service/src/interpreters/interpreter_select_v2.rs

@@ -18,7 +18,6 @@ use common_datavalues::DataSchemaRef;
 use common_exception::Result;
 
 use super::plan_schedulers::schedule_query_v2;
-use crate::clusters::ClusterHelper;
 use crate::interpreters::Interpreter;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;

src/query/service/src/optimizers/optimizer_scatters.rs

@@ -34,7 +34,6 @@ use common_legacy_planners::StagePlan;
 use common_legacy_planners::WindowFuncPlan;
 use enum_extract::let_extract;
 
-use crate::clusters::ClusterHelper;
 use crate::optimizers::Optimizer;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;

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/service/src/sql/planner/binder/insert.rs

@@ -21,7 +21,6 @@ use common_datavalues::DataSchemaRefExt;
 use common_exception::Result;
 use tracing::debug;
 
-use crate::clusters::ClusterHelper;
 use crate::sql::binder::Binder;
 use crate::sql::normalize_identifier;
 use crate::sql::optimizer::optimize;

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

@@ -23,7 +23,6 @@ use common_exception::Result;
 use parking_lot::RwLock;
 pub use plans::ScalarExpr;
 
-use crate::clusters::ClusterHelper;
 use crate::sql::optimizer::optimize;
 pub use crate::sql::planner::binder::BindContext;
 

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()?;