MDEV-36436 Acquire MDL on temporary name of persistent table during DDL

mysql-test/main/backup_lock.result

@@ -4,27 +4,27 @@
 InnoDB		0 transactions not purged
 BACKUP STAGE START;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_START	Backup lock		
 BACKUP STAGE FLUSH;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_FLUSH	Backup lock		
 BACKUP STAGE BLOCK_DDL;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_WAIT_DDL	Backup lock		
 BACKUP STAGE BLOCK_COMMIT;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_WAIT_COMMIT	Backup lock		
 BACKUP STAGE END;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 #
 # testing BACKUP STAGE LOCK's
@@ -47,13 +47,14 @@ alter table t1 add column (j int), algorithm copy, lock shared;
 connection con2;
 backup stage flush;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_DDL	Backup lock		
 MDL_BACKUP_FLUSH	Backup lock		
-MDL_SHARED_WRITE	Table metadata lock	test	t1
-MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_EXCLUSIVE	Table metadata lock	test	#sql-alter
 MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_SHARED_WRITE	Table metadata lock	test	t1
 SET STATEMENT max_statement_time=1 FOR backup stage block_ddl;
 ERROR 70100: Query execution was interrupted (max_statement_time exceeded)
 backup stage block_ddl;
@@ -84,13 +85,14 @@ connection con2;
 backup stage start;
 backup stage flush;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_ALTER_COPY	Backup lock		
 MDL_BACKUP_FLUSH	Backup lock		
-MDL_SHARED_WRITE	Table metadata lock	test	t1
-MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_EXCLUSIVE	Table metadata lock	test	#sql-alter
 MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_SHARED_WRITE	Table metadata lock	test	t1
 backup stage block_ddl;
 backup stage block_commit;
 connection default;
@@ -121,11 +123,11 @@ SET STATEMENT lock_wait_timeout=0 FOR SELECT * FROM t1;
 ERROR HY000: Lock wait timeout exceeded; try restarting transaction
 backup stage block_ddl;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_WAIT_DDL	Backup lock		
-MDL_SHARED_WRITE	Table metadata lock	test	t1
 MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+MDL_SHARED_WRITE	Table metadata lock	test	t1
 backup stage end;
 connection default;
 commit;
@@ -143,7 +145,7 @@ DROP TABLE t1;
 connection con2;
 connection con2;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
 MDL_BACKUP_WAIT_DDL	Backup lock		
 SELECT * FROM t1;

mysql-test/main/backup_lock.test

@@ -13,7 +13,7 @@
 --echo #
 
 let $mdl= LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
-WHERE TABLE_NAME NOT LIKE 'innodb_%_stats';
+WHERE TABLE_NAME NOT LIKE 'innodb_%_stats' ORDER BY LOCK_MODE;
 
 --source ../suite/innodb/include/wait_all_purged.inc
 
@@ -62,6 +62,7 @@ let $wait_condition=
     where state = "Waiting for table metadata lock";
 --source include/wait_condition.inc
 backup stage flush;
+--replace_regex /#sql-alter-[0-9a-f_\-]*/#sql-alter/
 eval SELECT $mdl;
 #
 # Do first test with max_statement_time, other tests later are done with
@@ -111,6 +112,7 @@ let $wait_condition=
 --source include/wait_condition.inc
 backup stage start;
 backup stage flush;
+--replace_regex /#sql-alter-[0-9a-f_\-]*/#sql-alter/
 eval SELECT $mdl;
 backup stage block_ddl;
 backup stage block_commit;

sql/sql_partition_admin.cc

@@ -716,6 +716,14 @@ bool Sql_cmd_alter_table_exchange_partition::
 
   DEBUG_SYNC(thd, "swap_partition_before_rename");
 
+  {
+    MDL_request mdl_request;
+    MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
+                     table_list->next_local->db.str,
+                     temp_name, MDL_EXCLUSIVE, MDL_TRANSACTION);
+    thd->mdl_context.acquire_lock(&mdl_request, 0);
+  }
+
   if (unlikely(exchange_name_with_ddl_log(thd, swap_file_name, part_file_name,
                                           temp_file_name, table_hton)))
     goto err;

sql/sql_table.cc

@@ -11588,6 +11588,19 @@ do_continue:;
   */
   alter_info->original_table= table;
 
+   {
+   /*
+     Create an exclusive lock on the temporary table name.
+     Needed by InnoDB storage engine to avoid MDL on source
+     table during purge.
+   */
+    MDL_request mdl_request;
+    MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
+                     alter_ctx.new_db.str, alter_ctx.tmp_name.str,
+                     MDL_EXCLUSIVE, MDL_TRANSACTION);
+    thd->mdl_context.acquire_lock(&mdl_request, 0);
+  }
+
   /*
     Create the .frm file for the new table. Storage engine table will not be
     created at this stage.
@@ -12160,6 +12173,16 @@ do_continue:;
       If we are changing to use another table handler, we don't
       have to do the rename as the table names will not interfer.
     */
+     /*
+       Create an exclusive lock on the backup table name.
+       Needed by InnoDB storage engine to avoid MDL conflicts
+       during backup table operations.
+     */
+    MDL_request mdl_request;
+    MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
+                     alter_ctx.db.str, backup_name.str,
+                     MDL_EXCLUSIVE, MDL_TRANSACTION);
+    thd->mdl_context.acquire_lock(&mdl_request, 0);
     if (mysql_rename_table(old_db_type, &alter_ctx.db, &alter_ctx.table_name,
                            &alter_ctx.db, &backup_name, &alter_ctx.id,
                            FN_TO_IS_TMP |

storage/innobase/dict/dict0dict.cc

@@ -519,59 +519,109 @@ void mdl_release(THD *thd, MDL_ticket *mdl) noexcept
     thd->mdl_context.release_lock(mdl);
 }
 
-/** Parse the table file name into table name and database name.
-@tparam        dict_frozen  whether the caller holds dict_sys.latch
-@param[in,out] db_name      database name buffer
-@param[in,out] tbl_name     table name buffer
-@param[out] db_name_len     database name length
-@param[out] tbl_name_len    table name length
-@return whether the table name is visible to SQL */
-template<bool dict_frozen>
-bool dict_table_t::parse_name(char (&db_name)[NAME_LEN + 1],
-                              char (&tbl_name)[NAME_LEN + 1],
-                              size_t *db_name_len, size_t *tbl_name_len) const
+void dict_table_t::parse_tbl_name(
+  const table_name_t &table_name,
+  char (&db_name)[NAME_LEN + 1],
+  char (&tbl_name)[NAME_LEN + 1],
+  size_t *db_name_len, size_t *tbl_name_len) noexcept
 {
   char db_buf[MAX_DATABASE_NAME_LEN + 1];
   char tbl_buf[MAX_TABLE_NAME_LEN + 1];
 
-  if (!dict_frozen)
-    dict_sys.freeze(SRW_LOCK_CALL); /* protect against renaming */
-  ut_ad(dict_sys.frozen());
-  const size_t db_len= name.dblen();
-  ut_ad(db_len <= MAX_DATABASE_NAME_LEN);
+  const size_t db_len= table_name.dblen();
+  const char* tbl_start= table_name.basename();
 
-  memcpy(db_buf, mdl_name.m_name, db_len);
-  db_buf[db_len]= 0;
+  if (db_len == 0 || !tbl_start)
+    return;
 
-  size_t tbl_len= strlen(mdl_name.m_name + db_len + 1);
-  const bool is_temp= mdl_name.is_temporary();
+  memcpy(db_buf, table_name.m_name, db_len);
+  db_buf[db_len]= '\0';
 
-  if (is_temp);
-  else if (const char *is_part= static_cast<const char*>
-           (memchr(mdl_name.m_name + db_len + 1, '#', tbl_len)))
-    tbl_len= static_cast<size_t>(is_part - &mdl_name.m_name[db_len + 1]);
+  size_t tbl_len= strlen(tbl_start);
+  const bool is_temp= table_name.is_temporary();
 
-  memcpy(tbl_buf, mdl_name.m_name + db_len + 1, tbl_len);
-  tbl_buf[tbl_len]= 0;
+  /* For partition tables, find the base table name by removing partition suffix */
+  if (const char *is_part = static_cast<const char*>
+      (memchr(tbl_start, '#', tbl_len)))
+  {
+    /* For temporary tables, we need to find the partition marker after the temp name */
+    if (is_temp)
+    {
+      /* Look for partition markers like #P# after the temporary table name */
+      const char *part_marker= strstr(tbl_start, "#P#");
+      if (part_marker)
+        tbl_len= static_cast<size_t>(part_marker - tbl_start);
+      else if (const char *vec_marker= strstr(tbl_start, "#i#"))
+        tbl_len= static_cast<size_t>(vec_marker - tbl_start);
+    }
+    else
+    {
+      /* For regular tables, use first # as partition boundary */
+      tbl_len= static_cast<size_t>(is_part - tbl_start);
+    }
+  }
 
-  if (!dict_frozen)
-    dict_sys.unfreeze();
+  memcpy(tbl_buf, tbl_start, tbl_len);
+  tbl_buf[tbl_len]= '\0';
 
+  /* Convert internal names to SQL names */
   *db_name_len= filename_to_tablename(db_buf, db_name,
                                       MAX_DATABASE_NAME_LEN + 1, true);
-
   if (is_temp)
-    return false;
-
+  {
+    memcpy(tbl_name, tbl_buf, tbl_len);
+    tbl_name[tbl_len]= '\0';
+    *tbl_name_len= tbl_len;
+    return;
+  }
   *tbl_name_len= filename_to_tablename(tbl_buf, tbl_name,
                                        MAX_TABLE_NAME_LEN + 1, true);
-  return true;
+  return;
 }
 
-template bool
+/** Parse the table file name into table name and database name.
+@tparam        dict_frozen  whether the caller holds dict_sys.latch
+@param[in,out] db_name      database name buffer
+@param[in,out] tbl_name     table name buffer
+@param[out] db_name_len     database name length
+@param[out] tbl_name_len    table name length
+@return DB_CORRUPTION if table name contains #sql-ib,
+@return DB_SUCCESS_LOCKED_REC if db_len == 0, DB_SUCCESS otherwise */
+template<bool dict_frozen>
+dberr_t dict_table_t::parse_name(char (&db_name)[NAME_LEN + 1],
+                                 char (&tbl_name)[NAME_LEN + 1],
+                                 size_t *db_name_len, size_t *tbl_name_len) const
+{
+  if (!dict_frozen)
+    dict_sys.freeze(SRW_LOCK_CALL); /* protect against renaming */
+  ut_ad(dict_sys.frozen());
+  parse_tbl_name(name, db_name, tbl_name, db_name_len, tbl_name_len);
+  if (!dict_frozen)
+    dict_sys.unfreeze();
+
+  /* Check for specific error conditions */
+  if (*db_name_len == 0)
+    return DB_SUCCESS_LOCKED_REC;
+
+  /* Inplace intermediate table should not generate any
+  undo logs. In that case, throw error */
+  if (strstr(tbl_name, "#sql-ib"))
+  {
+    sql_print_error("InnoDB: Trying to acquire MDL for inplace "
+                    "intermediate table #sql-ib");
+    return DB_CORRUPTION;
+  }
+  return DB_SUCCESS;
+}
+
+template dberr_t
 dict_table_t::parse_name<>(char(&)[NAME_LEN + 1], char(&)[NAME_LEN + 1],
                            size_t*, size_t*) const;
 
+template dberr_t
+dict_table_t::parse_name<true>(char(&)[NAME_LEN + 1], char(&)[NAME_LEN + 1],
+                               size_t*, size_t*) const;
+
 dict_table_t *dict_sys_t::acquire_temporary_table(table_id_t id) const noexcept
 {
   ut_ad(frozen());
@@ -631,10 +681,13 @@ dict_acquire_mdl_shared(dict_table_t *table,
   char tbl_buf[NAME_LEN + 1], tbl_buf1[NAME_LEN + 1];
   size_t db_len, tbl_len;
 
-  if (!table->parse_name<!trylock>(db_buf, tbl_buf, &db_len, &tbl_len))
-    /* The name of an intermediate table starts with #sql */
+  dberr_t err= table->parse_name<!trylock>(
+                 db_buf, tbl_buf, &db_len, &tbl_len);
+  if (err == DB_SUCCESS_LOCKED_REC)
     return table;
 
+  if (err == DB_CORRUPTION)
+    return nullptr;
 retry:
   ut_ad(!trylock == dict_sys.frozen());
 
@@ -701,10 +754,11 @@ dict_acquire_mdl_shared(dict_table_t *table,
     if (trylock)
       table->acquire();
 
-    if (!table->parse_name<true>(db_buf1, tbl_buf1, &db1_len, &tbl1_len))
+    dberr_t err=
+      table->parse_name<true>(db_buf1, tbl_buf1, &db1_len, &tbl1_len);
+    if (err == DB_SUCCESS_LOCKED_REC || err == DB_CORRUPTION)
     {
-      /* The table was renamed to #sql prefix.
-      Release MDL (if any) for the old name and return. */
+      /* Release MDL (if any) for the old name and return. */
       goto unlock_and_return_without_mdl;
     }
   }
@@ -1513,22 +1567,6 @@ dict_table_rename_in_cache(
 					       old_name_len))
 		->remove(*table, &dict_table_t::name_hash);
 
-	bool keep_mdl_name = !table->name.is_temporary();
-
-	if (!keep_mdl_name) {
-	} else if (const char* s = static_cast<const char*>
-		   (memchr(new_name.data(), '/', new_name.size()))) {
-		keep_mdl_name = new_name.end() - s >= 5
-			&& !memcmp(s, "/#sql", 5);
-	}
-
-	if (keep_mdl_name) {
-		/* Preserve the original table name for
-		dict_table_t::parse_name() and dict_acquire_mdl_shared(). */
-		table->mdl_name.m_name = mem_heap_strdup(table->heap,
-							 table->name.m_name);
-	}
-
 	if (new_name.size() > strlen(table->name.m_name)) {
 		/* We allocate MAX_FULL_NAME_LEN + 1 bytes here to avoid
 		memory fragmentation, we assume a repeated calls of
@@ -1541,10 +1579,6 @@ dict_table_rename_in_cache(
 	memcpy(table->name.m_name, new_name.data(), new_name.size());
 	table->name.m_name[new_name.size()] = '\0';
 
-	if (!keep_mdl_name) {
-		table->mdl_name.m_name = table->name.m_name;
-	}
-
 	/* Add table to hash table of tables */
 	ut_ad(!table->name_hash);
 	dict_table_t** after = reinterpret_cast<dict_table_t**>(

storage/innobase/dict/dict0mem.cc

@@ -148,7 +148,6 @@ dict_table_t *dict_table_t::create(const span<const char> &name,
   table->flags= static_cast<unsigned>(flags) & ((1U << DICT_TF_BITS) - 1);
   table->flags2= static_cast<unsigned>(flags2) & ((1U << DICT_TF2_BITS) - 1);
   table->name.m_name= mem_strdupl(name.data(), name.size());
-  table->mdl_name.m_name= table->name.m_name;
   table->is_system_db= dict_mem_table_is_system(table->name.m_name);
   table->space= space;
   table->space_id= space ? space->id : UINT32_MAX;