MDEV-31982 Remove check_db_name() from prepare_db_action()

- Adding a new class Lex_ident_db, to store normalized database names:
  lower-cased if lower-case-table-name says so,
  and checked to be a valid database name using Lex_ident_fs::check_db_name()

- Reusing the new class in parameters to functions:
    prepare_db_action()
    mysql_create_db()
    mysql_alter_db()
    mysql_rm_db()
    mysql_upgrade_db()

This change removed two old-style check_db_name() calls.

sql/lex_ident.h

@@ -47,6 +47,26 @@ class Lex_ident_fs: public LEX_CSTRING
 };
 
 
+/**
+  A normalized database name identifier.
+  - Converted to lower case if lower_case_table_names say so
+  - Checked to be a valid database name
+*/
+class Lex_ident_db: public Lex_ident_fs
+{
+public:
+  Lex_ident_db()
+   :Lex_ident_fs(NULL, 0)
+  { }
+  Lex_ident_db(const char *str, size_t length)
+   :Lex_ident_fs(str, length)
+  {
+    DBUG_SLOW_ASSERT(ok_for_lower_case_names());
+    DBUG_SLOW_ASSERT(!check_db_name());
+  }
+};
+
+
 /*
   A helper class to store temporary database names in a buffer.
   After constructing it's typically should be checked using
@@ -67,6 +87,13 @@ class DBNameBuffer: public CharBuffer<SAFE_NAME_LEN + MY_CS_MBMAXLEN>
   {
     copy_casedn(&my_charset_utf8mb3_general_ci, db, casedn);
   }
+  Lex_ident_db to_lex_ident_db_with_error() const
+  {
+    LEX_CSTRING tmp= to_lex_cstring();
+    if (Lex_ident_fs(tmp).check_db_name_with_error())
+      return Lex_ident_db();
+    return Lex_ident_db(tmp.str, tmp.length);
+  }
 };
 
 

sql/sql_class.cc

@@ -6078,6 +6078,26 @@ void THD::get_definer(LEX_USER *definer, bool role)
 }
 
 
+bool THD::check_slave_ignored_db_with_error(const Lex_ident_db &db) const
+{
+#ifdef HAVE_REPLICATION
+  if (slave_thread)
+  {
+    Rpl_filter *rpl_filter;
+    rpl_filter= system_thread_info.rpl_sql_info->rpl_filter;
+    if (!rpl_filter->db_ok(db.str) ||
+        !rpl_filter->db_ok_with_wild_table(db.str))
+    {
+      my_message(ER_SLAVE_IGNORED_TABLE,
+                 ER_THD(this, ER_SLAVE_IGNORED_TABLE), MYF(0));
+      return true;
+    }
+  }
+#endif
+  return false;
+}
+
+
 /**
   Mark transaction to rollback and mark error as fatal to a sub-statement.
 

sql/sql_class.h

@@ -5356,6 +5356,8 @@ class THD: public THD_count, /* this must be first */
   bool is_binlog_dump_thread();
 #endif
 
+  bool check_slave_ignored_db_with_error(const Lex_ident_db &db) const;
+
   /*
     Indicates if this thread is suspended due to awaiting an ACK from a
     replica. True if suspended, false otherwise.

sql/sql_db.h

@@ -20,12 +20,12 @@
 
 class THD;
 
-int mysql_create_db(THD *thd, const LEX_CSTRING *db, DDL_options_st options,
+int mysql_create_db(THD *thd, const Lex_ident_db &db, DDL_options_st options,
                     const Schema_specification_st *create);
-bool mysql_alter_db(THD *thd, const LEX_CSTRING *db,
+bool mysql_alter_db(THD *thd, const Lex_ident_db &db,
                     const Schema_specification_st *create);
-bool mysql_rm_db(THD *thd, const LEX_CSTRING *db, bool if_exists);
-bool mysql_upgrade_db(THD *thd, const LEX_CSTRING *old_db);
+bool mysql_rm_db(THD *thd, const Lex_ident_db &db, bool if_exists);
+bool mysql_upgrade_db(THD *thd, const Lex_ident_db &old_db);
 uint mysql_change_db(THD *thd, const LEX_CSTRING *new_db_name,
                      bool force_switch);
 

sql/sql_parse.cc

@@ -3178,13 +3178,8 @@ mysql_create_routine(THD *thd, LEX *lex)
   This can be done by testing thd->is_error().
 */
 static bool prepare_db_action(THD *thd, privilege_t want_access,
-                              LEX_CSTRING *dbname)
+                              const Lex_ident_db &dbname)
 {
-  if (check_db_name((LEX_STRING*)dbname))
-  {
-    my_error(ER_WRONG_DB_NAME, MYF(0), dbname->str);
-    return true;
-  }
   /*
     If in a slave thread :
     - CREATE DATABASE DB was certainly not preceded by USE DB.
@@ -3194,21 +3189,8 @@ static bool prepare_db_action(THD *thd, privilege_t want_access,
     do_db/ignore_db. And as this query involves no tables, tables_ok()
     was not called. So we have to check rules again here.
   */
-#ifdef HAVE_REPLICATION
-  if (thd->slave_thread)
-  {
-    Rpl_filter *rpl_filter;
-    rpl_filter= thd->system_thread_info.rpl_sql_info->rpl_filter;
-    if (!rpl_filter->db_ok(dbname->str) ||
-        !rpl_filter->db_ok_with_wild_table(dbname->str))
-    {
-      my_message(ER_SLAVE_IGNORED_TABLE,
-                 ER_THD(thd, ER_SLAVE_IGNORED_TABLE), MYF(0));
-      return true;
-    }
-  }
-#endif
-  return check_access(thd, want_access, dbname->str, NULL, NULL, 1, 0);
+  return thd->check_slave_ignored_db_with_error(dbname) ||
+         check_access(thd, want_access, dbname.str, NULL, NULL, 1, 0);
 }
 
 
@@ -3453,11 +3435,6 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
   TABLE_LIST *all_tables;
   /* most outer SELECT_LEX_UNIT of query */
   SELECT_LEX_UNIT *unit= &lex->unit;
-#ifdef HAVE_REPLICATION
-  /* have table map for update for multi-update statement (BUG#37051) */
-  /* */
-  Rpl_filter *rpl_filter;
-#endif
   DBUG_ENTER("mysql_execute_command");
 
   // check that we correctly marked first table for data insertion
@@ -4923,19 +4900,20 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
     break;
   case SQLCOM_CREATE_DB:
   {
-    if (prepare_db_action(thd, lex->create_info.or_replace() ?
-                          (CREATE_ACL | DROP_ACL) : CREATE_ACL,
-                          &lex->name))
-      break;
+    const DBNameBuffer dbbuf(lex->name, lower_case_table_names);
+    const Lex_ident_db db= dbbuf.to_lex_ident_db_with_error();
 
-    if ((res= lex->create_info.resolve_to_charset_collation_context(thd,
+    if (!db.str ||
+        prepare_db_action(thd, lex->create_info.or_replace() ?
+                          (CREATE_ACL | DROP_ACL) : CREATE_ACL,
+                          db) ||
+        (res= lex->create_info.resolve_to_charset_collation_context(thd,
                                  thd->charset_collation_context_create_db())))
       break;
 
-    WSREP_TO_ISOLATION_BEGIN(lex->name.str, NULL, NULL);
+    WSREP_TO_ISOLATION_BEGIN(db.str, NULL, NULL);
 
-    res= mysql_create_db(thd, &lex->name,
-                         lex->create_info, &lex->create_info);
+    res= mysql_create_db(thd, db, lex->create_info, &lex->create_info);
     break;
   }
   case SQLCOM_DROP_DB:
@@ -4943,44 +4921,33 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
     if (thd->variables.option_bits & OPTION_IF_EXISTS)
       lex->create_info.set(DDL_options_st::OPT_IF_EXISTS);
 
-    if (prepare_db_action(thd, DROP_ACL, &lex->name))
+    const DBNameBuffer dbbuf(lex->name, lower_case_table_names);
+    const Lex_ident_db db= dbbuf.to_lex_ident_db_with_error();
+
+    if (!db.str || prepare_db_action(thd, DROP_ACL, db))
       break;
 
-    WSREP_TO_ISOLATION_BEGIN(lex->name.str, NULL, NULL);
+    WSREP_TO_ISOLATION_BEGIN(db.str, NULL, NULL);
 
-    res= mysql_rm_db(thd, &lex->name, lex->if_exists());
+    res= mysql_rm_db(thd, db, lex->if_exists());
     break;
   }
   case SQLCOM_ALTER_DB_UPGRADE:
   {
-    LEX_CSTRING *db= &lex->name;
-#ifdef HAVE_REPLICATION
-    if (thd->slave_thread)
-    {
-      rpl_filter= thd->system_thread_info.rpl_sql_info->rpl_filter;
-      if (!rpl_filter->db_ok(db->str) ||
-          !rpl_filter->db_ok_with_wild_table(db->str))
-      {
-        res= 1;
-        my_message(ER_SLAVE_IGNORED_TABLE, ER_THD(thd, ER_SLAVE_IGNORED_TABLE), MYF(0));
-        break;
-      }
-    }
-#endif
-    if (check_db_name((LEX_STRING*) db))
-    {
-      my_error(ER_WRONG_DB_NAME, MYF(0), db->str);
-      break;
-    }
-    if (check_access(thd, ALTER_ACL, db->str, NULL, NULL, 1, 0) ||
-        check_access(thd, DROP_ACL, db->str, NULL, NULL, 1, 0) ||
-        check_access(thd, CREATE_ACL, db->str, NULL, NULL, 1, 0))
+    const DBNameBuffer dbbuf(lex->name, lower_case_table_names);
+    const Lex_ident_db db= dbbuf.to_lex_ident_db_with_error();
+
+    if (!db.str ||
+        thd->check_slave_ignored_db_with_error(db) ||
+        check_access(thd, ALTER_ACL, db.str, NULL, NULL, 1, 0) ||
+        check_access(thd, DROP_ACL, db.str, NULL, NULL, 1, 0) ||
+        check_access(thd, CREATE_ACL, db.str, NULL, NULL, 1, 0))
     {
       res= 1;
       break;
     }
 
-    WSREP_TO_ISOLATION_BEGIN(db->str, NULL, NULL);
+    WSREP_TO_ISOLATION_BEGIN(db.str, NULL, NULL);
 
     res= mysql_upgrade_db(thd, db);
     if (!res)
@@ -4989,15 +4956,16 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
   }
   case SQLCOM_ALTER_DB:
   {
-    LEX_CSTRING *db= &lex->name;
-    if (prepare_db_action(thd, ALTER_ACL, db))
-      break;
+    const DBNameBuffer dbbuf(lex->name, lower_case_table_names);
+    const Lex_ident_db db= dbbuf.to_lex_ident_db_with_error();
 
-    if ((res= lex->create_info.resolve_to_charset_collation_context(thd,
-                     thd->charset_collation_context_alter_db(lex->name.str))))
+    if (!db.str ||
+        prepare_db_action(thd, ALTER_ACL, db) ||
+        (res= lex->create_info.resolve_to_charset_collation_context(thd,
+                     thd->charset_collation_context_alter_db(db.str))))
       break;
 
-    WSREP_TO_ISOLATION_BEGIN(db->str, NULL, NULL);
+    WSREP_TO_ISOLATION_BEGIN(db.str, NULL, NULL);
 
     res= mysql_alter_db(thd, db, &lex->create_info);
     break;