MDEV-31978 Turn ok_for_lower_case_names() to a method in Lex_ident_fs

- Changing the global function ok_for_lower_case_names()
  into a method in class Lex_ident_fs.

- Changing a few functions/methods to get the database name
  as a "const LEX_CSTRING" instead of a "const char *".
  All these functions/methods use ok_for_lower_case_names()
  inside. This change helps to avoid new strlen() calls, and also
  removes a few old strlen() calls.

sql/events.cc

@@ -347,7 +347,7 @@ Events::create_event(THD *thd, Event_parse_data *parse_data)
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   if (lock_object_name(thd, MDL_key::EVENT,
-                       parse_data->dbname.str, parse_data->name.str))
+                       parse_data->dbname, parse_data->name))
     DBUG_RETURN(TRUE);
 
   if (check_db_dir_existence(parse_data->dbname.str))
@@ -475,7 +475,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   if (lock_object_name(thd, MDL_key::EVENT,
-                       parse_data->dbname.str, parse_data->name.str))
+                       parse_data->dbname, parse_data->name))
     DBUG_RETURN(TRUE);
 
   if (check_db_dir_existence(parse_data->dbname.str))
@@ -509,8 +509,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
     /*
      Acquire mdl exclusive lock on target database name.
     */
-    if (lock_object_name(thd, MDL_key::EVENT,
+    if (lock_object_name(thd, MDL_key::EVENT, *new_dbname, *new_name))
-                         new_dbname->str, new_name->str))
       DBUG_RETURN(TRUE);
 
     /* Check that the target database exists */
@@ -611,8 +610,7 @@ Events::drop_event(THD *thd, const LEX_CSTRING *dbname,
   */
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
-  if (lock_object_name(thd, MDL_key::EVENT,
+  if (lock_object_name(thd, MDL_key::EVENT, *dbname, *name))
-                       dbname->str, name->str))
     DBUG_RETURN(TRUE);
   /* On error conditions my_error() is called so no need to handle here */
   if (!(ret= db_repository->drop_event(thd, dbname, name, if_exists)))
@@ -648,14 +646,12 @@ wsrep_error_label:
 */
 
 void
-Events::drop_schema_events(THD *thd, const char *db)
+Events::drop_schema_events(THD *thd, const LEX_CSTRING &db_lex)
 {
-  const LEX_CSTRING db_lex= { db, strlen(db) };
-
   DBUG_ENTER("Events::drop_schema_events");
-  DBUG_PRINT("enter", ("dropping events from %s", db));
+  DBUG_PRINT("enter", ("dropping events from %s", db_lex.str));
 
-  DBUG_SLOW_ASSERT(ok_for_lower_case_names(db));
+  DBUG_SLOW_ASSERT(Lex_ident_fs(db_lex).ok_for_lower_case_names());
 
   /*
     Sic: no check if the scheduler is disabled or system tables

sql/events.h

@@ -119,7 +119,7 @@ public:
              bool if_exists);
 
   static void
-  drop_schema_events(THD *thd, const char *db);
+  drop_schema_events(THD *thd, const LEX_CSTRING &db);
 
   static bool
   show_create_event(THD *thd, const LEX_CSTRING *dbname,

sql/item_create.cc

@@ -2768,6 +2768,7 @@ Create_sp_func::create_with_db(THD *thd,
   sp_name *qname;
   const Sp_handler *sph= &sp_handler_function;
   Database_qualified_name pkgname(&null_clex_str, &null_clex_str);
+  DBUG_ASSERT(Lex_ident_fs(*db).ok_for_lower_case_names());
 
   if (unlikely(has_named_parameters(item_list)))
   {

sql/lex_ident.h

@@ -41,6 +41,9 @@ public:
                          bool disallow_path_chars);
   bool check_db_name() const;
   bool check_db_name_with_error() const;
+#ifndef DBUG_OFF
+  bool ok_for_lower_case_names() const;
+#endif
 };
 
 
@@ -55,7 +58,7 @@ public:
   (SYSTEM_CHARSET_MBMAXLEN bytes), so check_db_name() can still
   detect too long names even if the constructor cuts the data.
 */
-class DBNameBuffer: public CharBuffer<SAFE_NAME_LEN + SYSTEM_CHARSET_MBMAXLEN>
+class DBNameBuffer: public CharBuffer<SAFE_NAME_LEN + MY_CS_MBMAXLEN>
 {
 public:
   DBNameBuffer()

sql/lock.cc

@@ -935,14 +935,14 @@ bool lock_schema_name(THD *thd, const char *db)
 */
 
 bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
-                       const char *db, const char *name)
+                      const LEX_CSTRING &db, const LEX_CSTRING &name)
 {
   MDL_request_list mdl_requests;
   MDL_request global_request;
   MDL_request schema_request;
   MDL_request mdl_request;
 
-  DBUG_SLOW_ASSERT(ok_for_lower_case_names(db));
+  DBUG_SLOW_ASSERT(Lex_ident_fs(db).ok_for_lower_case_names());
 
   if (thd->locked_tables_mode)
   {
@@ -951,16 +951,16 @@ bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
     return TRUE;
   }
 
-  DBUG_ASSERT(name);
+  DBUG_ASSERT(name.str);
   DEBUG_SYNC(thd, "before_wait_locked_pname");
 
   if (thd->has_read_only_protection())
     return TRUE;
   MDL_REQUEST_INIT(&global_request, MDL_key::BACKUP, "", "", MDL_BACKUP_DDL,
                    MDL_STATEMENT);
-  MDL_REQUEST_INIT(&schema_request, MDL_key::SCHEMA, db, "",
+  MDL_REQUEST_INIT(&schema_request, MDL_key::SCHEMA, db.str, "",
                    MDL_INTENTION_EXCLUSIVE, MDL_TRANSACTION);
-  MDL_REQUEST_INIT(&mdl_request, mdl_type, db, name, MDL_EXCLUSIVE,
+  MDL_REQUEST_INIT(&mdl_request, mdl_type, db.str, name.str, MDL_EXCLUSIVE,
                    MDL_TRANSACTION);
 
   mdl_requests.push_front(&mdl_request);

sql/lock.h

@@ -39,7 +39,7 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a, MYSQL_LOCK *b, THD *thd= NULL);
 bool lock_schema_name(THD *thd, const char *db);
 /* Lock based on stored routine name */
 bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
-                      const char *db, const char *name);
+                      const LEX_CSTRING &db, const LEX_CSTRING &name);
 
 /* flags for get_lock_data */
 #define GET_LOCK_UNLOCK         0

sql/mdl.h

@@ -22,13 +22,13 @@
 #include <m_string.h>
 #include <mysql_com.h>
 #include <lf.h>
+#include "lex_ident.h"
 
 class THD;
 
 class MDL_context;
 class MDL_lock;
 class MDL_ticket;
-bool  ok_for_lower_case_names(const char *name);
 
 typedef unsigned short mdl_bitmap_t;
 #define MDL_BIT(A) static_cast<mdl_bitmap_t>(1U << A)
@@ -435,7 +435,9 @@ public:
                                           NAME_LEN) - m_ptr + 1);
     m_hash_value= my_hash_sort(&my_charset_bin, (uchar*) m_ptr + 1,
                                m_length - 1);
-    DBUG_SLOW_ASSERT(mdl_namespace_arg == USER_LOCK || ok_for_lower_case_names(db));
+    DBUG_SLOW_ASSERT(mdl_namespace_arg == USER_LOCK ||
+                     Lex_ident_fs(db, m_db_name_length).
+                       ok_for_lower_case_names());
   }
   void mdl_key_init(const MDL_key *rhs)
   {

sql/sp.cc

@@ -1277,7 +1277,7 @@ Sp_handler::sp_create_routine(THD *thd, const sp_head *sp) const
   retstr.set_charset(system_charset_info);
 
   /* Grab an exclusive MDL lock. */
-  if (lock_object_name(thd, mdl_type, sp->m_db.str, sp->m_name.str))
+  if (lock_object_name(thd, mdl_type, sp->m_db, sp->m_name))
   {
     my_error(ER_BAD_DB_ERROR, MYF(0), sp->m_db.str);
     DBUG_RETURN(TRUE);
@@ -1647,7 +1647,7 @@ Sp_handler::sp_drop_routine(THD *thd,
   MDL_key::enum_mdl_namespace mdl_type= get_mdl_type();
 
   /* Grab an exclusive MDL lock. */
-  if (lock_object_name(thd, mdl_type, name->m_db.str, name->m_name.str))
+  if (lock_object_name(thd, mdl_type, name->m_db, name->m_name))
     DBUG_RETURN(SP_DELETE_ROW_FAILED);
 
   if (!(table= open_proc_table_for_update(thd)))
@@ -1694,7 +1694,7 @@ Sp_handler::sp_update_routine(THD *thd, const Database_qualified_name *name,
   MDL_key::enum_mdl_namespace mdl_type= get_mdl_type();
 
   /* Grab an exclusive MDL lock. */
-  if (lock_object_name(thd, mdl_type, name->m_db.str, name->m_name.str))
+  if (lock_object_name(thd, mdl_type, name->m_db, name->m_name))
     DBUG_RETURN(SP_OPEN_TABLE_FAILED);
 
   if (!(table= open_proc_table_for_update(thd)))
@@ -1795,7 +1795,7 @@ public:
          cases.
  */
 
-bool lock_db_routines(THD *thd, const char *db)
+bool lock_db_routines(THD *thd, const LEX_CSTRING &db)
 {
   TABLE *table;
   uint key_len;
@@ -1804,7 +1804,7 @@ bool lock_db_routines(THD *thd, const char *db)
   uchar keybuf[MAX_KEY_LENGTH];
   DBUG_ENTER("lock_db_routines");
 
-  DBUG_SLOW_ASSERT(ok_for_lower_case_names(db));
+  DBUG_SLOW_ASSERT(Lex_ident_fs(db).ok_for_lower_case_names());
 
   start_new_trans new_trans(thd);
 
@@ -1827,7 +1827,7 @@ bool lock_db_routines(THD *thd, const char *db)
     DBUG_RETURN(thd->is_error() || thd->killed);
   }
 
-  table->field[MYSQL_PROC_FIELD_DB]->store(db, strlen(db), system_charset_info);
+  table->field[MYSQL_PROC_FIELD_DB]->store(db, system_charset_info);
   key_len= table->key_info->key_part[0].store_length;
   table->field[MYSQL_PROC_FIELD_DB]->get_key_image(keybuf, key_len, Field::itRAW);
   int nxtres= table->file->ha_index_init(0, 1);
@@ -1853,7 +1853,7 @@ bool lock_db_routines(THD *thd, const char *db)
                                                  sp_type);
       if (!sph)
         sph= &sp_handler_procedure;
-      MDL_REQUEST_INIT(mdl_request, sph->get_mdl_type(), db, sp_name,
+      MDL_REQUEST_INIT(mdl_request, sph->get_mdl_type(), db.str, sp_name,
                         MDL_EXCLUSIVE, MDL_TRANSACTION);
       mdl_requests.push_front(mdl_request);
     } while (! (nxtres= table->file->ha_index_next_same(table->record[0], keybuf, key_len)));
@@ -1870,7 +1870,7 @@ bool lock_db_routines(THD *thd, const char *db)
   /* We should already hold a global IX lock and a schema X lock. */
   DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
                                              MDL_BACKUP_DDL) &&
-              thd->mdl_context.is_lock_owner(MDL_key::SCHEMA, db, "",
+              thd->mdl_context.is_lock_owner(MDL_key::SCHEMA, db.str, "",
                                              MDL_EXCLUSIVE));
   DBUG_RETURN(thd->mdl_context.acquire_locks(&mdl_requests,
                                              thd->variables.lock_wait_timeout));
@@ -1889,23 +1889,22 @@ error:
 */
 
 int
-sp_drop_db_routines(THD *thd, const char *db)
+sp_drop_db_routines(THD *thd, const LEX_CSTRING &db)
 {
   TABLE *table;
   int ret;
   uint key_len;
   MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   uchar keybuf[MAX_KEY_LENGTH];
-  size_t db_length= strlen(db);
   Sql_mode_instant_remove smir(thd, MODE_PAD_CHAR_TO_FULL_LENGTH); // see below
   DBUG_ENTER("sp_drop_db_routines");
-  DBUG_PRINT("enter", ("db: %s", db));
+  DBUG_PRINT("enter", ("db: %s", db.str));
 
   ret= SP_OPEN_TABLE_FAILED;
   if (!(table= open_proc_table_for_update(thd)))
     goto err;
 
-  table->field[MYSQL_PROC_FIELD_DB]->store(db, db_length, system_charset_info);
+  table->field[MYSQL_PROC_FIELD_DB]->store(db, system_charset_info);
   key_len= table->key_info->key_part[0].store_length;
   table->field[MYSQL_PROC_FIELD_DB]->get_key_image(keybuf, key_len, Field::itRAW);
 
@@ -1933,7 +1932,8 @@ sp_drop_db_routines(THD *thd, const char *db)
 
         enum_sp_type sp_type= (enum_sp_type) table->field[MYSQL_PROC_MYSQL_TYPE]->ptr[0];
         /* Drop statistics for this stored program from performance schema. */
-        MYSQL_DROP_SP(sp_type, db, static_cast<uint>(db_length), name->ptr(), name->length());
+        MYSQL_DROP_SP(sp_type, db.str, static_cast<uint>(db.length),
+                      name->ptr(), name->length());
 #endif
       }
       else

sql/sp.h

@@ -582,7 +582,7 @@ enum
 
 /* Drop all routines in database 'db' */
 int
-sp_drop_db_routines(THD *thd, const char *db);
+sp_drop_db_routines(THD *thd, const LEX_CSTRING &db);
 
 /**
    Acquires exclusive metadata lock on all stored routines in the
@@ -594,7 +594,7 @@ sp_drop_db_routines(THD *thd, const char *db);
    @retval  false  Success
    @retval  true   Failure
  */
-bool lock_db_routines(THD *thd, const char *db);
+bool lock_db_routines(THD *thd, const LEX_CSTRING &db);
 
 /**
   Structure that represents element in the set of stored routines

sql/sql_cache.cc

@@ -2351,13 +2351,13 @@ void Query_cache::invalidate(THD *thd, const char *key, size_t  key_length,
    Remove all cached queries that uses the given database.
 */
 
-void Query_cache::invalidate(THD *thd, const char *db)
+void Query_cache::invalidate(THD *thd, const LEX_CSTRING &db)
 {
   DBUG_ENTER("Query_cache::invalidate (db)");
   if (is_disabled())
     DBUG_VOID_RETURN;
 
-  DBUG_SLOW_ASSERT(ok_for_lower_case_names(db));
+  DBUG_SLOW_ASSERT(Lex_ident_fs(db).ok_for_lower_case_names());
 
   bool restart= FALSE;
   /*
@@ -2377,7 +2377,7 @@ void Query_cache::invalidate(THD *thd, const char *db)
         {
           Query_cache_block *next= table_block->next;
           Query_cache_table *table = table_block->table();
-          if (strcmp(table->db(),db) == 0)
+          if (strcmp(table->db(), db.str) == 0)
           {
             Query_cache_block_table *list_root= table_block->table(0);
             invalidate_query_block_list(list_root);

sql/sql_cache.h

@@ -485,7 +485,7 @@ protected:
 		  my_bool using_transactions);
 
   /* Remove all queries that uses any of the tables in following database */
-  void invalidate(THD *thd, const char *db);
+  void invalidate(THD *thd, const LEX_CSTRING &db);
 
   /* Remove all queries that uses any of the listed following table */
   void invalidate_by_MyISAM_filename(const char *filename);

sql/sql_class.h

@@ -7844,7 +7844,7 @@ public:
             (int) m_db.length, (m_db.length ? m_db.str : ""),
             dot, ".",
             (int) m_name.length, m_name.str);
-    DBUG_SLOW_ASSERT(ok_for_lower_case_names(m_db.str));
+    DBUG_SLOW_ASSERT(Lex_ident_fs(m_db).ok_for_lower_case_names());
     return false;
   }
 

sql/sql_db.cc

@@ -1003,14 +1003,14 @@ void drop_database_objects(THD *thd, const LEX_CSTRING *path,
 
   debug_crash_here("ddl_log_drop_before_drop_db_routines");
 
-  query_cache_invalidate1(thd, db->str);
+  query_cache_invalidate1(thd, *db);
 
   if (!rm_mysql_schema)
   {
     tmp_disable_binlog(thd);
-    (void) sp_drop_db_routines(thd, db->str); /* @todo Do not ignore errors */
+    (void) sp_drop_db_routines(thd, *db); /* @todo Do not ignore errors */
 #ifdef HAVE_EVENT_SCHEDULER
-    Events::drop_schema_events(thd, db->str);
+    Events::drop_schema_events(thd, *db);
 #endif
     reenable_binlog(thd);
   }
@@ -1093,7 +1093,7 @@ mysql_rm_db_internal(THD *thd, const LEX_CSTRING *db, bool if_exists,
   /* Lock all tables and stored routines about to be dropped. */
   if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
                        0) ||
-      lock_db_routines(thd, rm_db.str))
+      lock_db_routines(thd, rm_db))
     goto exit;
 
   if (!rm_mysql_schema)

sql/table.cc

@@ -5176,15 +5176,12 @@ uint calculate_key_len(TABLE *table, uint key, const uchar *buf,
 
   This is supposed to be used only inside DBUG_ASSERT()
 */
-bool ok_for_lower_case_names(const char *name)
+bool Lex_ident_fs::ok_for_lower_case_names() const
 {
-  if (!lower_case_table_names || !name)
+  if (!lower_case_table_names || !str)
     return true;
-
+  DBNameBuffer buf(*this, lower_case_table_names);
-  char buf[SAFE_NAME_LEN];
+  return cmp(*this, buf.to_lex_cstring()) == 0;
-  strmake_buf(buf, name);
-  my_casedn_str(files_charset_info, buf);
-  return strcmp(name, buf) == 0;
 }
 #endif
 

sql/table.h

@@ -3365,7 +3365,6 @@ static inline void dbug_tmp_restore_column_maps(MY_BITMAP **read_set,
 #endif
 }
 
-bool ok_for_lower_case_names(const char *names);
 
 enum get_table_share_flags {
   GTS_TABLE                = 1,