MDEV-36493 Atomic CREATE OR REPLACE ... SELECT blocks InnoDB purge

dict_table_t::parse_tbl_name() : Extracts database and table
components from full table names

ddl_log_acquire_mdl(): Does take MDL on given table
  - DELETE/REPLACE: single table MDL acquisition
  - RENAME: source and destination table MDL protection
  - ALTER: comprehensive MDL on original, intermediate
    and final tables
  - EXCHANGE: primary, source, and temporary table MDL coverage

MDL acquistion on temporary tables in mysql_alter_table(),
crash recovery for DDL log entry.

innodb_mdl_acquire() for temporary tables in
commit_inplace_alter_table()

innodb_mdl_check() to check MDL ownership verifiction in
rename_table() and delete_table(). In innodb_rename_table(),
it does check for source and destination table MDL check

- InnoDB purge thread now takes MDL on table name directly
instead of relying on mdl_name

Author: Thirunarayanan

mysql-test/main/backup_lock.result

@@ -54,6 +54,7 @@ MDL_BACKUP_FLUSH	Backup lock
 MDL_SHARED_WRITE	Table metadata lock	test	t1
 MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
 MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+MDL_EXCLUSIVE	Table metadata lock	test	#sql-alter
 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;
@@ -90,6 +91,7 @@ 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	
 backup stage block_ddl;
 backup stage block_commit;

mysql-test/main/backup_lock.test

@@ -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/ddl_log.cc

@@ -85,7 +85,6 @@ uchar ddl_log_file_magic[]=
 { (uchar) 254, (uchar) 254, (uchar) 11, (uchar) 2 };
 
 /* Action names for ddl_log_action_code */
-
 const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
 {
   "Unknown", "partitioning delete", "partitioning rename",
@@ -96,6 +95,136 @@ const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
   "delete tmp file", "create trigger", "alter table", "store query"
 };
 
+/**
+  Acquire MDL exclusive lock on a single table name.
+  @param  thd         MySQL thread handle
+  @param  db_name     Database name
+  @param  tbl_name    Table name
+*/
+static void ddl_log_acquire_table_mdl(
+    THD *thd, const char *db_name, const char *tbl_name)
+{
+  if (!db_name || !tbl_name || !*db_name || !*tbl_name)
+    return;
+  MDL_request request;
+  MDL_REQUEST_INIT(&request, MDL_key::TABLE, db_name, tbl_name,
+                   MDL_EXCLUSIVE, MDL_TRANSACTION);
+
+  thd->mdl_context.acquire_lock(&request, 0);
+}
+
+/**
+  Conditionally acquire MDL based on database name availability.
+  If db_name is empty, treat table_name as full path and parse it.
+  Otherwise, use db_name and table_name directly.
+  @param  thd         MySQL thread handle
+  @param  db_name     Database name (may be empty)
+  @param  table_name  Table name or full path
+*/
+static void ddl_log_acquire_table_mdl(
+    THD *thd, const LEX_CSTRING &db_name, const LEX_CSTRING &table_name)
+{
+  if (db_name.length == 0)
+  {
+    /* If db is empty, treat table_name as full path and parse it */
+    const char *full_name= table_name.str;
+    if (!full_name || !*full_name)
+      return;
+
+    /* Skip leading ./ if present */
+    if (full_name[0] == '.' && full_name[1] == '/')
+      full_name+= 2;
+
+    /* Find database/table separator (either '/' or '.') */
+    const char *separator= strchr(full_name, '/');
+    if (!separator)
+      separator= strchr(full_name, '.');
+
+    if (!separator)
+      return;
+
+    const char *db_start= full_name;
+    const char *tbl_start= separator + 1;
+    size_t db_len= separator - db_start;
+    size_t tbl_len= strlen(tbl_start);
+
+    if (db_len == 0 || tbl_len == 0 || db_len > NAME_LEN || tbl_len > NAME_LEN)
+      return;
+
+    /* Handle partition table names - look for partition markers like #P# */
+    const char *part_marker= strstr(tbl_start, "#P#");
+    if (part_marker)
+      tbl_len= part_marker - tbl_start;
+
+    /* Copy and null-terminate names */
+    char parsed_db_name[NAME_LEN + 1];
+    char parsed_tbl_name[NAME_LEN + 1];
+    memcpy(parsed_db_name, db_start, db_len);
+    parsed_db_name[db_len]= '\0';
+    memcpy(parsed_tbl_name, tbl_start, tbl_len);
+    parsed_tbl_name[tbl_len]= '\0';
+
+    ddl_log_acquire_table_mdl(thd, parsed_db_name, parsed_tbl_name);
+  }
+  else
+  {
+    /* Use db and table names directly */
+    ddl_log_acquire_table_mdl(thd, db_name.str, table_name.str);
+  }
+}
+
+/**
+  Acquire MDL exclusive locks for DDL log operations based on action type.
+  @param  thd             MySQL thread handle
+  @param  ddl_log_entry   DDL log entry containing action type and table names
+*/
+static void ddl_log_acquire_mdl(THD *thd, DDL_LOG_ENTRY *ddl_log_entry)
+{
+  if (memcmp(ddl_log_entry->handler_name.str, "InnoDB",
+             ddl_log_entry->handler_name.length))
+    return;
+
+  switch (ddl_log_entry->action_type) {
+  case DDL_LOG_DELETE_ACTION:
+  case DDL_LOG_REPLACE_ACTION:
+    /* Acquire MDL on table being deleted/replaced */
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->name);
+    break;
+
+  case DDL_LOG_RENAME_ACTION:
+  case DDL_LOG_RENAME_TABLE_ACTION:
+    /* Acquire MDL on both source and destination tables for rename */
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->from_db,
+                              ddl_log_entry->from_name);
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->name);
+    break;
+
+  case DDL_LOG_ALTER_TABLE_ACTION:
+    /* Acquire MDL on all tables involved in alter table operations */
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->name);
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->from_db,
+                              ddl_log_entry->from_name);
+    ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->extra_name);
+    break;
+
+  case DDL_LOG_EXCHANGE_ACTION:
+    {
+      /* Acquire MDL on primary table */
+      ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->name);
+
+      /* Acquire MDL on source table for exchange */
+      ddl_log_acquire_table_mdl(thd, ddl_log_entry->from_db,
+                                ddl_log_entry->from_name);
+
+      ddl_log_acquire_table_mdl(thd, ddl_log_entry->db, ddl_log_entry->tmp_name);
+      break;
+    }
+  default:
+    /* No MDL acquisition needed for other action types */
+    break;
+  }
+}
+
 /* Number of phases per entry */
 const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
 {
@@ -1446,6 +1575,10 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     invers_fn_flags|= FN_FROM_IS_TMP;
   }
 
+  if (!frm_action)
+    ddl_log_acquire_mdl(thd, ddl_log_entry);
+
+  /* Acquire MDL for table operations */
   switch (ddl_log_entry->action_type) {
   case DDL_LOG_REPLACE_ACTION:
   case DDL_LOG_DELETE_ACTION:

sql/sql_partition_admin.cc

@@ -661,6 +661,14 @@ bool Sql_cmd_alter_table_exchange_partition::
                        table_list->next_local->db.str,
                        temp_name, "", FN_IS_TMP);
 
+  {
+    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(!(part_elem=
                  part_table->part_info->get_part_elem(partition_name,
                                                       part_file_name +

sql/sql_table.cc

@@ -11929,6 +11929,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.
@@ -12501,6 +12514,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,6 +519,64 @@ void mdl_release(THD *thd, MDL_ticket *mdl) noexcept
     thd->mdl_context.release_lock(mdl);
 }
 
+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];
+
+  const size_t db_len= table_name.dblen();
+  const char* tbl_start= table_name.basename();
+
+  if (db_len == 0 || !tbl_start)
+    return;
+
+  memcpy(db_buf, table_name.m_name, db_len);
+  db_buf[db_len]= '\0';
+
+  size_t tbl_len= strlen(tbl_start);
+  const bool is_temp= table_name.is_temporary() || table_name.is_create_or_replace();
+
+  /* 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
+    {
+      /* For regular tables, use first # as partition boundary */
+      tbl_len= static_cast<size_t>(is_part - tbl_start);
+    }
+  }
+
+  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)
+  {
+    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;
+}
+
 /** 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
@@ -531,47 +589,23 @@ 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
 {
-  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);
-
-  memcpy(db_buf, mdl_name.m_name, db_len);
-  db_buf[db_len]= 0;
-
-  size_t tbl_len= strlen(mdl_name.m_name + db_len + 1);
-  const bool is_temp= mdl_name.is_temporary();
-
-  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]);
-
-  memcpy(tbl_buf, mdl_name.m_name + db_len + 1, tbl_len);
-  tbl_buf[tbl_len]= 0;
-
+  parse_tbl_name(name, db_name, tbl_name, db_name_len, tbl_name_len);
   if (!dict_frozen)
     dict_sys.unfreeze();
-
-  *db_name_len= filename_to_tablename(db_buf, db_name,
-                                      MAX_DATABASE_NAME_LEN + 1, true);
-
-  if (is_temp)
-    return false;
-
-  *tbl_name_len= filename_to_tablename(tbl_buf, tbl_name,
-                                       MAX_TABLE_NAME_LEN + 1, true);
   return true;
 }
 
 template bool
 dict_table_t::parse_name<>(char(&)[NAME_LEN + 1], char(&)[NAME_LEN + 1],
                            size_t*, size_t*) const;
 
+template bool
+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());