fix: match system schema names case-insensitively [Backport release/v1.2] (#9041)

fix: match system schema names case-insensitively (#9040)

* fix: match system schema names case-insensitively

Database names that arrive over a protocol (the MySQL handshake and
COM_INIT_DB, the Postgres startup parameter, the HTTP `db` parameter, the
gRPC dbname header) never reach the SQL parser, which is what lowercases
unquoted identifiers. Since #8062 stopped lowercasing them wholesale,
connecting to `INFORMATION_SCHEMA` in any spelling but the canonical one
fails with "Unknown database" -- including the `USE <db>` that a MySQL
client turns into COM_INIT_DB.

Fold only system schema names to their canonical spelling, so user schema
names keep the case they were created with. `is_reserved_schema_name` uses
the same match, otherwise a quoted `CREATE DATABASE "INFORMATION_SCHEMA"`
creates a schema shadowed by the system one.



* refactor: hoist system schema names into a const



---------


(cherry picked from commit fa794fae7a)

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
Co-authored-by: dennis zhuang <killme2008@gmail.com>

src/catalog/src/lib.rs

@@ -17,7 +17,7 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use api::v1::CreateTableExpr;
-use common_catalog::consts::{INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME};
+use common_catalog::consts::system_schema_name;
 use futures::future::BoxFuture;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
@@ -125,7 +125,10 @@ pub trait CatalogManager: Send + Sync {
         // We need this rather than use schema_exists directly because `pg_catalog` is
         // only visible via postgres protocol. So if we don't check, a mysql client may
         // create a schema named `pg_catalog` which is somehow malformed.
-        schema == INFORMATION_SCHEMA_NAME || schema == PG_CATALOG_NAME
+        //
+        // Case-insensitive, otherwise `CREATE DATABASE "INFORMATION_SCHEMA"` keeps its
+        // case and creates a schema shadowed by the system one.
+        system_schema_name(schema).is_some()
     }
 }
 

src/catalog/src/memory/manager.rs

@@ -450,6 +450,15 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_reserved_schema_name_ignores_case() {
+        let catalog = MemoryCatalogManager::with_default_setup();
+
+        assert!(catalog.is_reserved_schema_name("INFORMATION_SCHEMA"));
+        assert!(catalog.is_reserved_schema_name(&DEFAULT_PRIVATE_SCHEMA_NAME.to_uppercase()));
+        assert!(!catalog.is_reserved_schema_name("my_information_schema"));
+    }
+
     #[tokio::test]
     async fn test_new_memory_catalog_list() {
         let catalog_list = new_memory_catalog_manager().unwrap();

src/common/catalog/src/consts.rs

@@ -144,6 +144,23 @@ pub const SEMANTIC_TYPE_PRIMARY_KEY: &str = "TAG";
 pub const SEMANTIC_TYPE_FIELD: &str = "FIELD";
 pub const SEMANTIC_TYPE_TIME_INDEX: &str = "TIMESTAMP";
 
+const SYSTEM_SCHEMA_NAMES: [&str; 3] = [
+    INFORMATION_SCHEMA_NAME,
+    PG_CATALOG_NAME,
+    DEFAULT_PRIVATE_SCHEMA_NAME,
+];
+
+/// Returns the canonical name of the system schema `schema` refers to, ignoring ASCII
+/// case, or `None` if it is not one.
+///
+/// Only system schemas are matched case-insensitively, as MySQL does; user schema names
+/// keep the case they were created with.
+pub fn system_schema_name(schema: &str) -> Option<&'static str> {
+    SYSTEM_SCHEMA_NAMES
+        .into_iter()
+        .find(|name| schema.eq_ignore_ascii_case(name))
+}
+
 pub fn is_readonly_schema(schema: &str) -> bool {
     matches!(schema, INFORMATION_SCHEMA_NAME)
 }

src/common/catalog/src/lib.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use consts::DEFAULT_CATALOG_NAME;
+use consts::{DEFAULT_CATALOG_NAME, system_schema_name};
 
 pub mod consts;
 
@@ -74,14 +74,25 @@ pub fn parse_catalog_and_schema_from_db_string(db: &str) -> (String, String) {
 pub fn parse_optional_catalog_and_schema_from_db_string(db: &str) -> (Option<String>, String) {
     let parts = db.splitn(2, '-').collect::<Vec<&str>>();
     if parts.len() == 2 {
-        (Some(parts[0].to_string()), parts[1].to_string())
+        (
+            Some(parts[0].to_string()),
+            canonicalize_schema_name(parts[1]),
+        )
     } else {
-        (None, db.to_string())
+        (None, canonicalize_schema_name(db))
     }
 }
 
+/// A database name taken from a protocol never reaches the SQL parser, which is what
+/// lowercases unquoted identifiers, so system schemas are folded here instead.
+fn canonicalize_schema_name(schema: &str) -> String {
+    system_schema_name(schema).unwrap_or(schema).to_string()
+}
+
 #[cfg(test)]
 mod tests {
+    use consts::{INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME};
+
     use super::*;
 
     #[test]
@@ -127,4 +138,17 @@ mod tests {
             parse_optional_catalog_and_schema_from_db_string("catalog-schema1-schema2")
         );
     }
+
+    #[test]
+    fn test_parse_system_schema_ignores_case() {
+        assert_eq!(
+            (None, INFORMATION_SCHEMA_NAME.to_string()),
+            parse_optional_catalog_and_schema_from_db_string("INFORMATION_SCHEMA")
+        );
+
+        assert_eq!(
+            (Some("CATALOG".to_string()), PG_CATALOG_NAME.to_string()),
+            parse_optional_catalog_and_schema_from_db_string("CATALOG-Pg_Catalog")
+        );
+    }
 }

tests-integration/tests/mysql.rs

@@ -122,3 +122,29 @@ OK packet (rows affected: 0)
     guard.remove_all().await;
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_mysql_connect_system_schema_ignores_case() -> sqlx::Result<()> {
+    let (mut guard, server) = setup_mysql_server(
+        StorageType::File,
+        "test_mysql_connect_system_schema_ignores_case",
+    )
+    .await;
+    let addr = server.bind_addr().unwrap();
+
+    // The handshake database goes through the same `on_init` as `COM_INIT_DB`, which is
+    // what a MySQL client turns `USE <db>` into. Neither reaches the SQL parser.
+    let mut conn = MySqlConnection::connect(&format!("mysql://{addr}/INFORMATION_SCHEMA")).await?;
+
+    // Unqualified: resolves only if the session really switched to the system schema.
+    let schema: String =
+        sqlx::query("select table_schema from tables where table_name = 'columns'")
+            .fetch_one(&mut conn)
+            .await?
+            .get(0);
+    assert_eq!(schema, "information_schema");
+
+    let _ = server.shutdown().await;
+    guard.remove_all().await;
+    Ok(())
+}