MDEV-32081 Remove my_casedn_str() from get_canonical_filename()

- Moving get_canonical_filename() from a public function to a method in handler.
- Adding a helper method is_canonical_filename() to handler.
- Adding helper methods left(), substr(), starts_with() to Lex_cstring.
- Adding helper methods is_sane(), buffer_overlaps(),
  max_data_size() to CharBuffer.
- Adding append_casedn() to CharBuffer. It implements the main functionality
  that replaces the being removed my_casedn_str() call.
- Adding a class Table_path_buffer,
  a descendant of CharBuffer with size FN_REFLEN.
- Changing get_canonical_filename() to get a pointer to Table_path_buffer
  instead just a pointer to char.
- Changing the data type of the "path" parameter and the return type of
  get_canonical_filename() from char* to Lex_cstring.

sql/char_buffer.h

@@ -33,7 +33,19 @@ class CharBuffer
 {
   char m_buff[buff_sz + 1 /* one extra byte for '\0' */];
   size_t m_length;
+  bool is_sane() const
+  {
+    return m_length <= buff_sz; // One byte is still left for '\0'
+  }
+  bool buffer_overlaps(const LEX_CSTRING &str) const
+  {
+    return str.str + str.length >= m_buff && str.str <= m_buff + sizeof(m_buff);
+  }
 public:
+  constexpr size_t max_data_size() const
+  {
+    return buff_sz; // The maximum data size, without the trailing '\0' byte.
+  }
   CharBuffer()
    :m_length(0)
   {
@@ -41,6 +53,7 @@ class CharBuffer
   }
   CharBuffer<buff_sz> & copy_bin(const LEX_CSTRING &str)
   {
+    DBUG_ASSERT(!buffer_overlaps(str));
     m_length= MY_MIN(buff_sz, str.length);
     memcpy(m_buff, str.str, m_length);
     m_buff[m_length]= '\0';
@@ -48,12 +61,13 @@ class CharBuffer
   }
   CharBuffer<buff_sz> & copy_casedn(CHARSET_INFO *cs, const LEX_CSTRING &str)
   {
+    DBUG_ASSERT(!buffer_overlaps(str));
     m_length= cs->cset->casedn(cs, str.str, str.length, m_buff, buff_sz);
     /*
       casedn() never writes outsize of buff_sz (unless a bug in casedn()),
       so it's safe to write '\0' at the position m_length:
     */
-    DBUG_ASSERT(m_length <= buff_sz);
+    DBUG_ASSERT(is_sane());
     m_buff[m_length]= '\0';
     return *this;
   }
@@ -63,6 +77,19 @@ class CharBuffer
     casedn ? copy_casedn(cs, str) : copy_bin(str);
     return *this;
   }
+  // Append a string with casedn conversion
+  CharBuffer<buff_sz> & append_casedn(CHARSET_INFO *cs, const LEX_CSTRING &str)
+  {
+    DBUG_ASSERT(is_sane());
+    DBUG_ASSERT(!buffer_overlaps(str));
+    size_t casedn_length= cs->casedn(str.str, str.length,
+                                     m_buff + m_length, buff_sz - m_length);
+    m_length+= casedn_length;
+    DBUG_ASSERT(is_sane());
+    m_buff[m_length]= '\0';
+    return *this;
+  }
+
   LEX_CSTRING to_lex_cstring() const
   {
     return LEX_CSTRING{m_buff, m_length};

sql/ha_partition.cc

@@ -99,19 +99,19 @@ int ha_partition::notify_tabledef_changed(LEX_CSTRING *db,
                                           LEX_CUSTRING *frm,
                                           LEX_CUSTRING *version)
 {
-  char from_buff[FN_REFLEN + 1], from_lc_buff[FN_REFLEN + 1];
-  const char *from_path, *name_buffer_ptr, *from;
+  char from_buff[FN_REFLEN + 1];
+  Table_path_buffer from_lc_buff;
+  const char *from_path, *name_buffer_ptr;
   int res= 0;
   handler **file= m_file;
   DBUG_ENTER("ha_partition::notify_tabledef_changed");
 
-  from= table->s->normalized_path.str;
-
   /* setup m_name_buffer_ptr */
   if (read_par_file(table->s->normalized_path.str))
     DBUG_RETURN(1);
 
-  from_path= get_canonical_filename(*file, from, from_lc_buff);
+  from_path= file[0]->get_canonical_filename(table->s->normalized_path,
+                                             &from_lc_buff).str;
   name_buffer_ptr= m_name_buffer_ptr;
   do
   {
@@ -769,7 +769,8 @@ int ha_partition::create(const char *name, TABLE *table_arg,
 {
   int error;
   THD *thd= ha_thd();
-  char name_buff[FN_REFLEN + 1], name_lc_buff[FN_REFLEN];
+  char name_buff[FN_REFLEN + 1];
+  Table_path_buffer name_lc_buff;
   char *name_buffer_ptr;
   const char *path;
   uint i;
@@ -819,7 +820,8 @@ int ha_partition::create(const char *name, TABLE *table_arg,
     The appended #P#<partname>[#SP#<subpartname>] will remain in current case.
     Using the first partitions handler, since mixing handlers is not allowed.
   */
-  path= get_canonical_filename(*file, name, name_lc_buff);
+  path= file[0]->get_canonical_filename(Lex_cstring_strlen(name),
+                                        &name_lc_buff).str;
   for (i= 0; i < m_part_info->num_parts; i++)
   {
     part_elem= part_it++;
@@ -911,8 +913,8 @@ int ha_partition::drop_partitions(const char *path)
     Assert that it works without HA_FILE_BASED and lower_case_table_name = 2.
     We use m_file[0] as long as all partitions have the same storage engine.
   */
-  DBUG_ASSERT(!strcmp(path, get_canonical_filename(m_file[0], path,
-                                                   part_name_buff)));
+  DBUG_ASSERT(m_file[0]->is_canonical_filename(Lex_cstring_strlen(path)));
+
   do
   {
     partition_element *part_elem= part_it++;
@@ -1016,8 +1018,7 @@ int ha_partition::rename_partitions(const char *path)
     Assert that it works without HA_FILE_BASED and lower_case_table_name = 2.
     We use m_file[0] as long as all partitions have the same storage engine.
   */
-  DBUG_ASSERT(!strcmp(path, get_canonical_filename(m_file[0], path,
-                                                   norm_name_buff)));
+  DBUG_ASSERT(m_file[0]->is_canonical_filename(Lex_cstring_strlen(path)));
 
   DEBUG_SYNC(ha_thd(), "before_rename_partitions");
   if (temp_partitions)
@@ -1855,8 +1856,8 @@ int ha_partition::change_partitions(HA_CREATE_INFO *create_info,
     Assert that it works without HA_FILE_BASED and lower_case_table_name = 2.
     We use m_file[0] as long as all partitions have the same storage engine.
   */
-  DBUG_ASSERT(!strcmp(path, get_canonical_filename(m_file[0], path,
-                                                   part_name_buff)));
+  DBUG_ASSERT(m_file[0]->is_canonical_filename(Lex_cstring_strlen(path)));
+
   m_reorged_parts= 0;
   if (!m_part_info->is_sub_partitioned())
     num_subparts= 1;
@@ -2448,8 +2449,8 @@ uint ha_partition::del_ren_table(const char *from, const char *to)
 {
   int save_error= 0;
   int error;
-  char from_buff[FN_REFLEN + 1], to_buff[FN_REFLEN + 1],
-       from_lc_buff[FN_REFLEN], to_lc_buff[FN_REFLEN];
+  char from_buff[FN_REFLEN + 1], to_buff[FN_REFLEN + 1];
+  Table_path_buffer from_lc_buff, to_lc_buff;
   char *name_buffer_ptr;
   const char *from_path;
   const char *to_path= NULL;
@@ -2489,9 +2490,11 @@ uint ha_partition::del_ren_table(const char *from, const char *to)
     The appended #P#<partname>[#SP#<subpartname>] will remain in current case.
     Using the first partitions handler, since mixing handlers is not allowed.
   */
-  from_path= get_canonical_filename(*file, from, from_lc_buff);
+  from_path= file[0]->get_canonical_filename(Lex_cstring_strlen(from),
+                                             &from_lc_buff).str;
   if (to != NULL)
-    to_path= get_canonical_filename(*file, to, to_lc_buff);
+    to_path= file[0]->get_canonical_filename(Lex_cstring_strlen(to),
+                                             &to_lc_buff).str;
   do
   {
     if (unlikely((error= create_partition_name(from_buff, sizeof(from_buff),

sql/handler.cc

@@ -565,7 +565,7 @@ int ha_drop_table(THD *thd, handlerton *hton, const char *path)
 
 static int hton_drop_table(handlerton *hton, const char *path)
 {
-  char tmp_path[FN_REFLEN];
+  Table_path_buffer tmp_path;
   handler *file= get_new_handler(nullptr, current_thd->mem_root, hton);
   if (!file)
   {
@@ -575,7 +575,7 @@ static int hton_drop_table(handlerton *hton, const char *path)
     */
     return my_errno == ENOMEM ? ENOMEM : ENOENT;
   }
-  path= get_canonical_filename(file, path, tmp_path);
+  path= file->get_canonical_filename(Lex_cstring_strlen(path), &tmp_path).str;
   int error= file->delete_table(path);
   delete file;
   return error;
@@ -3118,9 +3118,8 @@ bool ha_flush_logs()
 /**
   @brief make canonical filename
 
-  @param[in]  file     table handler
   @param[in]  path     original path
-  @param[out] tmp_path buffer for canonized path
+  @param[out] buff     buffer for canonized path
 
   @details Lower case db name and table name path parts for
            non file based tables when lower_case_table_names
@@ -3128,8 +3127,7 @@ bool ha_flush_logs()
            Filesystem path prefix (mysql_data_home or tmpdir)
            is left intact.
 
-  @note tmp_path may be left intact if no conversion was
-        performed.
+  @note buff may be left intact if no conversion was performed.
 
   @retval canonized path
 
@@ -3137,29 +3135,40 @@ bool ha_flush_logs()
         gets built. Convert this function to something like
         ASSERT_CANONICAL_FILENAME.
 */
-const char *get_canonical_filename(handler *file, const char *path,
-                                   char *tmp_path)
+
+Lex_cstring handler::get_canonical_filename(const Lex_cstring &path,
+                                            Table_path_buffer *buff) const
 {
+  /* Ensure that table handler get path in lower case */
   uint i;
-  if (!file->needs_lower_case_filenames())
+  if (!needs_lower_case_filenames())
     return path;
 
   for (i= 0; i <= mysql_tmpdir_list.max; i++)
   {
-    if (is_prefix(path, mysql_tmpdir_list.list[i]))
+    if (is_prefix(path.str, mysql_tmpdir_list.list[i]))
       return path;
   }
-
-  /* Ensure that table handler get path in lower case */
-  if (tmp_path != path)
-    strmov(tmp_path, path);
-
   /*
     we only should turn into lowercase database/table part
-    so start the process after homedirectory
+    so start lower-casing after homedirectory
+  */
+  DBUG_ASSERT(path.starts_with({mysql_data_home, mysql_data_home_len}));
+  /*
+    QQ: important for upgrade from MySQL-5.0 with --lower-case-table-names=2
+
+    In case if both the table name and the database name are encoded
+    using tablename_to_filename(), the it's ok to lower-case the entire
+    "/db/table" part using files_charset_info.
+
+    Otherwise, in case if either of the table name or the database name
+    starts with '#mysql50#', it's probably not correct to lower-case using
+    files_charset_info. Shoudn't we lower-case '#mysql50#name' using
+    character_set_filesystem instead?
   */
-  my_casedn_str(files_charset_info, tmp_path + mysql_data_home_len);
-  return tmp_path;
+  return buff->set_casedn(path.left(mysql_data_home_len),
+                          files_charset_info,
+                          path.substr(mysql_data_home_len)).to_lex_cstring();
 }
 
 
@@ -6092,7 +6101,7 @@ int ha_create_table(THD *thd, const char *path, const char *db,
 {
   int error= 1;
   TABLE table;
-  char name_buff[FN_REFLEN];
+  Table_path_buffer name_buff;
   const char *name;
   TABLE_SHARE share;
   Abort_on_warning_instant_set old_abort_on_warning(thd, 0);
@@ -6131,7 +6140,7 @@ int ha_create_table(THD *thd, const char *path, const char *db,
 
   update_create_info_from_table(create_info, &table);
 
-  name= get_canonical_filename(table.file, share.path.str, name_buff);
+  name= table.file->get_canonical_filename(share.path, &name_buff).str;
 
   error= table.file->ha_create(name, &table, create_info);
 

sql/handler.h

@@ -3113,6 +3113,39 @@ enum class Compare_keys : uint32_t
 };
 
 
+/*
+  This class stores a table file name in the format:
+     homedir/db/table
+  where db and table use tablename_to_filename() compatible encoding.
+*/
+class Table_path_buffer: public CharBuffer<FN_REFLEN>
+{
+public:
+  Table_path_buffer()
+  { }
+  /**
+    Make a lower-cased path for a table.
+    @homedir      - home directory, get copied to the buffer as is
+                    (without lower-casing)
+    @db_and_table - the database and the table name part in the format
+                    "/db/table", can be in arbitrary letter case. It gets
+                    converted to lower case during copying to the buffer.
+                    The "db" and "table" parts can be prefixed with '#myql50#'.
+
+    Makes the path in the format "homedir/db/table".
+  */
+  Table_path_buffer & set_casedn(const Lex_cstring &homedir,
+                                 CHARSET_INFO *db_and_table_charset,
+                                 const Lex_cstring &db_and_table)
+  {
+    DBUG_ASSERT(homedir.length + db_and_table.length <= max_data_size());
+    copy_bin(homedir);
+    append_casedn(db_and_table_charset, db_and_table);
+    return *this;
+  }
+};
+
+
 /**
   The handler class is the interface for dynamically loadable
   storage engines. Do not add ifdefs and take care when adding or
@@ -5441,11 +5474,21 @@ class handler :public Sql_alloc
   file system) and the storage is not HA_FILE_BASED, we need to provide
   a lowercase file name for the engine.
 */
-  inline bool needs_lower_case_filenames()
+  inline bool needs_lower_case_filenames() const
   {
     return (lower_case_table_names == 2 && !(ha_table_flags() & HA_FILE_BASED));
   }
 
+  Lex_cstring get_canonical_filename(const Lex_cstring &path,
+                                     Table_path_buffer *tmp_path)
+                                     const;
+
+  bool is_canonical_filename(const LEX_CSTRING &path) const
+  {
+    Table_path_buffer cpath;
+    return !strcmp(path.str, get_canonical_filename(path, &cpath).str);
+  }
+
   bool log_not_redoable_operation(const char *operation);
 
 protected:
@@ -5627,8 +5670,7 @@ void trans_register_ha(THD *thd, bool all, handlerton *ht,
 #define trans_need_2pc(thd, all)                   ((total_ha_2pc > 1) && \
         !((all ? &thd->transaction.all : &thd->transaction.stmt)->no_2pc))
 
-const char *get_canonical_filename(handler *file, const char *path,
-                                   char *tmp_path);
+
 void commit_checkpoint_notify_ha(void *cookie);
 
 inline const LEX_CSTRING *table_case_name(HA_CREATE_INFO *info, const LEX_CSTRING *name)

sql/lex_string.h

@@ -50,6 +50,31 @@ class Lex_cstring : public LEX_CSTRING
     str= _str;
     length= _len;
   }
+  /*
+    Return the "n" leftmost bytes if this[0] is longer than "n" bytes,
+    or return this[0] itself otherwise.
+  */
+  Lex_cstring left(size_t n) const
+  {
+    return Lex_cstring(str, MY_MIN(length, n));
+  }
+  /*
+    If this[0] is shorter than "pos" bytes, then return an empty string.
+    Otherwise, return a substring of this[0] starting from
+    the byte position "pos" until the end.
+  */
+  Lex_cstring substr(size_t pos) const
+  {
+    return length <= pos ? Lex_cstring(str + length, (size_t) 0) :
+                           Lex_cstring(str + pos, length - pos);
+  }
+  // Check if a prefix of this[0] is equal to "rhs".
+  bool starts_with(const LEX_CSTRING &rhs) const
+  {
+    DBUG_ASSERT(str);
+    DBUG_ASSERT(rhs.str);
+    return length >= rhs.length && !memcmp(str, rhs.str, rhs.length);
+  }
 };