MDEV-14410 - Assertion `table->pos_in_locked_tables == __null ||

             table->pos_in_locked_tables->table == table'
             failed in mark_used_tables_as_free_for_reuse

Assertion failure can be triggered by some DDL executed under LOCK TABLES
that holds lock for DDL target table multiple times (either explicitly or
implcitly).

When closing all table instances for given table (e.g. when preparing for
table removal during CREATE OR REPLACE), only one instance was removed
from m_locked_tables list.

Later we attempt to re-insert one of the instances in mysql_create_table()/
add_back_last_deleted_lock(), which wasn't actually removed. This leads
to m_locks_tables corruption, specifically loss of all following elements.

Then UNLOCK TABLE won't reset some table instances properly (specifically
pos_in_locked_tables), since they're not present in m_locked_tables.

Eventually such table instance gets released to table cache and then
re-used by subsequent statement, which triggers this assertion failure.

4 files changed, 39 insertions, 1 deletions

diff --git a/mysql-test/r/create_or_replace.result b/mysql-test/r/create_or_replace.result
index 0d171f9f87a..646183ebcb1 100644
--- a/mysql-test/r/create_or_replace.result
+++ b/mysql-test/r/create_or_replace.result
@@ -473,3 +473,18 @@ ERROR HY000: Table 't1' was not locked with LOCK TABLES
 UNLOCK TABLES;
 DROP FUNCTION f1;
 DROP TABLE t1;
+#
+# MDEV-14410 - Assertion `table->pos_in_locked_tables == __null ||
+#              table->pos_in_locked_tables->table == table' failed in
+#              mark_used_tables_as_free_for_reuse
+#
+CREATE TABLE t1 (a INT);
+CREATE TABLE t2 (b INT);
+CREATE TABLE t3 (c INT);
+CREATE TRIGGER tr1 BEFORE INSERT ON t3 FOR EACH ROW INSERT INTO t1 VALUES ();
+CREATE TRIGGER tr2 BEFORE INSERT ON t2 FOR EACH ROW INSERT INTO t3 SELECT * FROM t1;
+LOCK TABLE t1 WRITE, t2 WRITE;
+CREATE OR REPLACE TABLE t1 (i INT);
+UNLOCK TABLES;
+INSERT INTO t2 VALUES (1);
+DROP TABLE t1, t2, t3;
diff --git a/mysql-test/t/create_or_replace.test b/mysql-test/t/create_or_replace.test
index 5ebb3031be3..8c8ed1c9886 100644
--- a/mysql-test/t/create_or_replace.test
+++ b/mysql-test/t/create_or_replace.test
@@ -423,3 +423,24 @@ UNLOCK TABLES;
 
 DROP FUNCTION f1;
 DROP TABLE t1;
+
+
+--echo #
+--echo # MDEV-14410 - Assertion `table->pos_in_locked_tables == __null ||
+--echo #              table->pos_in_locked_tables->table == table' failed in
+--echo #              mark_used_tables_as_free_for_reuse
+--echo #
+CREATE TABLE t1 (a INT);
+CREATE TABLE t2 (b INT);
+CREATE TABLE t3 (c INT);
+
+CREATE TRIGGER tr1 BEFORE INSERT ON t3 FOR EACH ROW INSERT INTO t1 VALUES ();
+CREATE TRIGGER tr2 BEFORE INSERT ON t2 FOR EACH ROW INSERT INTO t3 SELECT * FROM t1;
+
+LOCK TABLE t1 WRITE, t2 WRITE;
+CREATE OR REPLACE TABLE t1 (i INT);
+UNLOCK TABLES;
+INSERT INTO t2 VALUES (1);
+
+# Cleanup
+DROP TABLE t1, t2, t3;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 32ce3c3a793..1a17c8e39d5 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -806,6 +806,7 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
   uint key_length= share->table_cache_key.length;
   const char *db= key;
   const char *table_name= db + share->db.length + 1;
+  bool remove_from_locked_tables= extra != HA_EXTRA_NOT_USED;
 
   memcpy(key, share->table_cache_key.str, key_length);
 
@@ -819,7 +820,7 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
     {
       thd->locked_tables_list.unlink_from_list(thd,
                                                table->pos_in_locked_tables,
-                                               extra != HA_EXTRA_NOT_USED);
+                                               remove_from_locked_tables);
       /* Inform handler that there is a drop table or a rename going on */
       if (extra != HA_EXTRA_NOT_USED && table->db_stat)
       {
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index bdb7914c32b..f13d7183a99 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -376,6 +376,7 @@ bool tc_release_table(TABLE *table)
 {
   DBUG_ASSERT(table->in_use);
   DBUG_ASSERT(table->file);
+  DBUG_ASSERT(!table->pos_in_locked_tables);
 
   if (table->needs_reopen() || tc_records() > tc_size)
   {