MDEV-25905: Assertion table2==NULL in dict_sys_t::add()

In commit 49e2c8f (MDEV-25743)
we made dict_sys_t::find() incompatible with the rest of the
table name hash table operations in case the table name contains
non-ASCII octets (using a compatibility mode that facilitates the
upgrade into the MySQL 5.0 filename-safe encoding) and the target
platform implements signed char.

ut_fold_string(): Remove; replace with my_crc32c(). This also makes
table name hash value calculations independent on whether char
is unsigned or signed.

Author: dr-m

extra/mariabackup/xtrabackup.cc

@@ -2414,7 +2414,8 @@ find_filter_in_hashtable(
 )
 {
 	xb_filter_entry_t* found = NULL;
-	HASH_SEARCH(name_hash, table, ut_fold_string(name),
+	const ulint fold = my_crc32c(0, name, strlen(name));
+	HASH_SEARCH(name_hash, table, fold,
 		    xb_filter_entry_t*,
 		    found, (void) 0,
 		    !strcmp(found->name, name));
@@ -3874,22 +3875,16 @@ new hash table */
 static
 xb_filter_entry_t*
 xb_add_filter(
-/*========================*/
 	const char*	name,	/*!< in: name of table/database */
 	hash_table_t*	hash)	/*!< in/out: hash to insert into */
 {
-	xb_filter_entry_t*	entry;
-
-	entry = xb_new_filter_entry(name);
+	xb_filter_entry_t* entry = xb_new_filter_entry(name);
 
 	if (UNIV_UNLIKELY(!hash->array)) {
 		hash->create(1000);
 	}
-	HASH_INSERT(xb_filter_entry_t,
-		name_hash, hash,
-		ut_fold_string(entry->name),
-		entry);
-
+	const ulint fold = my_crc32c(0, entry->name, strlen(entry->name));
+	HASH_INSERT(xb_filter_entry_t, name_hash, hash, fold, entry);
 	return entry;
 }
 
@@ -3943,8 +3938,9 @@ xb_register_filter_entry(
 		dbname[p - name] = 0;
 
 		if (databases_hash && databases_hash->array) {
+			const ulint fold = my_crc32c(0, dbname, p - name);
 			HASH_SEARCH(name_hash, databases_hash,
-					ut_fold_string(dbname),
+					fold,
 					xb_filter_entry_t*,
 					db_entry, (void) 0,
 					!strcmp(db_entry->name, dbname));
@@ -4153,9 +4149,10 @@ xb_filter_hash_free(hash_table_t* hash)
 
 			table = static_cast<xb_filter_entry_t *>
 				(HASH_GET_NEXT(name_hash, prev_table));
-
+			const ulint fold = my_crc32c(0, prev_table->name,
+						     strlen(prev_table->name));
 			HASH_DELETE(xb_filter_entry_t, name_hash, hash,
-				ut_fold_string(prev_table->name), prev_table);
+				    fold, prev_table);
 			free(prev_table);
 		}
 	}
@@ -5049,15 +5046,17 @@ xb_delta_open_matching_space(
 		return file;
 	}
 
+	const size_t len = strlen(dest_space_name);
 	/* remember space name for further reference */
 	table = static_cast<xb_filter_entry_t *>
 		(malloc(sizeof(xb_filter_entry_t) +
-			strlen(dest_space_name) + 1));
+			len + 1));
 
 	table->name = ((char*)table) + sizeof(xb_filter_entry_t);
-	strcpy(table->name, dest_space_name);
+	memcpy(table->name, dest_space_name, len + 1);
+	const ulint fold = my_crc32c(0, dest_space_name, len);
 	HASH_INSERT(xb_filter_entry_t, name_hash, &inc_dir_tables_hash,
-			ut_fold_string(table->name), table);
+		    fold, table);
 
 	mysql_mutex_lock(&fil_system.mutex);
 	fil_space = fil_space_get_by_name(dest_space_name);
@@ -5458,8 +5457,10 @@ static ibool prepare_handle_new_files(const char *data_home_dir,
 			(malloc(sizeof(xb_filter_entry_t) + table_name.size() + 1));
 		table->name = ((char*)table) + sizeof(xb_filter_entry_t);
 		strcpy(table->name, table_name.c_str());
+		const ulint fold = my_crc32c(0, table->name,
+					     table_name.size());
 		HASH_INSERT(xb_filter_entry_t, name_hash, &inc_dir_tables_hash,
-				ut_fold_string(table->name), table);
+			    fold, table);
 	}
 
 	return TRUE;
@@ -5482,9 +5483,11 @@ rm_if_not_found(
 
 	snprintf(name, FN_REFLEN, "%s/%s", db_name, file_name);
 	/* Truncate ".ibd" */
-	name[strlen(name) - 4] = '\0';
+	const size_t len = strlen(name) - 4;
+	name[len] = '\0';
+	const ulint fold = my_crc32c(0, name, len);
 
-	HASH_SEARCH(name_hash, &inc_dir_tables_hash, ut_fold_string(name),
+	HASH_SEARCH(name_hash, &inc_dir_tables_hash, fold,
 		    xb_filter_entry_t*,
 		    table, (void) 0,
 		    !strcmp(table->name, name));

mysql-test/suite/innodb/r/dropdb.result

@@ -1,3 +1,5 @@
+SET NAMES utf8;
+call mtr.add_suppression("Invalid .old.. table or database name");
 #
 # Bug #19929435 DROP DATABASE HANGS WITH MALFORMED TABLE
 #
@@ -6,4 +8,5 @@ create database `b`;
 use `b`;
 create table `#mysql50#q.q` select 1;
 ERROR 42000: Incorrect table name '#mysql50#q.q'
+create table `#mysql50#q·q` select 1;
 drop database `b`;

mysql-test/suite/innodb/t/dropdb.test

@@ -1,5 +1,8 @@
 --source include/have_innodb.inc
 
+SET NAMES utf8;
+call mtr.add_suppression("Invalid .old.. table or database name");
+
 --echo #
 --echo # Bug #19929435 DROP DATABASE HANGS WITH MALFORMED TABLE
 --echo #
@@ -9,4 +12,5 @@ create database `b`;
 use `b`;
 --error ER_WRONG_TABLE_NAME
 create table `#mysql50#q.q` select 1;
+create table `#mysql50#q·q` select 1;
 drop database `b`;

storage/innobase/dict/dict0dict.cc

@@ -1201,7 +1201,8 @@ inline void dict_sys_t::add(dict_table_t* table)
 {
 	ut_ad(!find(table));
 
-	ulint fold = ut_fold_string(table->name.m_name);
+	ulint fold = my_crc32c(0, table->name.m_name,
+			       strlen(table->name.m_name));
 
 	table->autoinc_mutex.init();
 	table->lock_mutex_init();
@@ -1545,10 +1546,11 @@ dict_table_rename_in_cache(
 	dict_sys.assert_locked();
 
 	/* store the old/current name to an automatic variable */
-	ut_a(strlen(table->name.m_name) < sizeof old_name);
+	const size_t old_name_len = strlen(table->name.m_name);
+	ut_a(old_name_len < sizeof old_name);
 	strcpy(old_name, table->name.m_name);
 
-	fold = ut_fold_string(new_name);
+	fold = my_crc32c(0, new_name, strlen(new_name));
 
 	/* Look for a table with the same name: error if such exists */
 	dict_table_t*	table2;
@@ -1560,7 +1562,7 @@ dict_table_rename_in_cache(
 			table2 = (dict_table_t*) -1;
 		} );
 	if (table2) {
-		ib::error() << "Cannot rename table '" << old_name
+		ib::error() << "Cannot rename table '" << table->name
 			<< "' to '" << new_name << "' since the"
 			" dictionary cache already contains '" << new_name << "'.";
 		return(DB_ERROR);
@@ -1574,7 +1576,7 @@ dict_table_rename_in_cache(
 
 	/* Remove table from the hash tables of tables */
 	HASH_DELETE(dict_table_t, name_hash, &dict_sys.table_hash,
-		    ut_fold_string(old_name), table);
+		    my_crc32c(0, table->name.m_name, old_name_len), table);
 
 	const bool keep_mdl_name = dict_table_t::is_temporary_name(new_name)
 		&& !table->name.is_temporary();
@@ -1922,7 +1924,9 @@ void dict_sys_t::remove(dict_table_t* table, bool lru, bool keep)
 	/* Remove table from the hash tables of tables */
 
 	HASH_DELETE(dict_table_t, name_hash, &table_hash,
-		    ut_fold_string(table->name.m_name), table);
+		    my_crc32c(0, table->name.m_name,
+			      strlen(table->name.m_name)),
+		    table);
 
 	hash_table_t* id_hash = table->is_temporary()
 		? &temp_id_hash : &table_id_hash;
@@ -4663,7 +4667,7 @@ void dict_sys_t::resize()
        table= UT_LIST_GET_NEXT(table_LRU, table))
   {
     ut_ad(!table->is_temporary());
-    ulint fold= ut_fold_string(table->name.m_name);
+    ulint fold= my_crc32c(0, table->name.m_name, strlen(table->name.m_name));
     ulint id_fold= ut_fold_ull(table->id);
 
     HASH_INSERT(dict_table_t, name_hash, &table_hash, fold, table);
@@ -4673,7 +4677,7 @@ void dict_sys_t::resize()
   for (dict_table_t *table = UT_LIST_GET_FIRST(table_non_LRU); table;
        table= UT_LIST_GET_NEXT(table_LRU, table))
   {
-    ulint fold= ut_fold_string(table->name.m_name);
+    ulint fold= my_crc32c(0, table->name.m_name, strlen(table->name.m_name));
     ulint id_fold= ut_fold_ull(table->id);
 
     HASH_INSERT(dict_table_t, name_hash, &table_hash, fold, table);

storage/innobase/include/dict0dict.h

@@ -32,6 +32,7 @@ Created 1/8/1996 Heikki Tuuri
 #include "dict0mem.h"
 #include "fsp0fsp.h"
 #include "srw_lock.h"
+#include <my_sys.h>
 #include <deque>
 
 class MDL_ticket;
@@ -1478,7 +1479,7 @@ class dict_sys_t
   {
     mysql_mutex_assert_owner(&mutex);
     dict_table_t *table;
-    ulint fold = ut_fold_ull(id);
+    ulint fold= ut_fold_ull(id);
     HASH_SEARCH(id_hash, &table_id_hash, fold, dict_table_t*, table,
                 ut_ad(table->cached), table->id == id);
     DBUG_ASSERT(!table || !table->is_temporary());
@@ -1617,8 +1618,7 @@ class dict_sys_t
     assert_locked();
     for (dict_table_t *table= static_cast<dict_table_t*>
          (HASH_GET_FIRST(&table_hash, table_hash.calc_hash
-                         (ut_fold_binary(reinterpret_cast<const byte*>
-                                         (name.data()), name.size()))));
+                         (my_crc32c(0, name.data(), name.size()))));
          table; table= table->name_hash)
       if (strlen(table->name.m_name) == name.size() &&
           !memcmp(table->name.m_name, name.data(), name.size()))

storage/innobase/include/ut0rnd.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, 2020, MariaDB Corporation.
+Copyright (c) 2019, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -90,15 +90,6 @@ ut_fold_ull(
 /*========*/
 	ib_uint64_t	d)	/*!< in: 64-bit integer */
 	MY_ATTRIBUTE((const));
-/*************************************************************//**
-Folds a character string ending in the null character.
-@return folded value */
-UNIV_INLINE
-ulint
-ut_fold_string(
-/*===========*/
-	const char*	str)	/*!< in: null-terminated string */
-	MY_ATTRIBUTE((warn_unused_result));
 /***********************************************************//**
 Looks for a prime number slightly greater than the given argument.
 The prime is chosen so that it is not near any power of 2.

storage/innobase/include/ut0rnd.ic

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -59,28 +59,6 @@ ut_fold_ull(
 	return(ut_fold_ulint_pair((ulint) d & ULINT32_MASK,
 				  (ulint) (d >> 32)));
 }
-
-/*************************************************************//**
-Folds a character string ending in the null character.
-@return folded value */
-UNIV_INLINE
-ulint
-ut_fold_string(
-/*===========*/
-	const char*	str)	/*!< in: null-terminated string */
-{
-	ulint	fold = 0;
-
-	ut_ad(str);
-
-	while (*str != '\0') {
-		fold = ut_fold_ulint_pair(fold, (ulint)(*str));
-		str++;
-	}
-
-	return(fold);
-}
-
 #endif /* !UNIV_INNOCHECKSUM */
 
 /*************************************************************//**

tpool/aio_liburing.cc

@@ -162,12 +162,10 @@ class aio_uring final : public tpool::aio
 
       io_uring_cqe_seen(&aio->uring_, cqe);
 
-      if (res == -EAGAIN) {
-        // If we need to resubmit the IO operation, but the ring is full,
-        // then just go the same path as for any other error codes.
-        if (!aio->submit_io(iocb))
-          continue;
-      }
+      // If we need to resubmit the IO operation, but the ring is full,
+      // we will follow the same path as for any other error codes.
+      if (res == -EAGAIN && !aio->submit_io(iocb))
+        continue;
 
       iocb->m_internal_task.m_func= iocb->m_callback;
       iocb->m_internal_task.m_arg= iocb;