squash! 7b54e04

MDEV-36501 EITS data is lost after failed attempt to CREATE OR REPLACE
           table
MDEV-36493 Atomic CREATE OR REPLACE ... SELECT blocks InnoDB purge

Other things:
- EITS data is preserved if create or replace fails if
  drop_before_create_or_replace=OFF. If ON, then create or replace
  will drop EITS before the drop of the original table (as before).

Author: montywi

mysql-test/main/create_or_replace.result

@@ -1173,5 +1173,65 @@ SELECT 1;
 1
 DROP TABLE t1;
 #
+# MDEV-36501 EITS data is lost after failed attempt to
+# CREATE OR REPLACE table
+#
+create table t1 (a int primary key);
+insert into t1 values (1),(2);
+analyze table t1 persistent for all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select count(*) from mysql.table_stats where table_name = 't1';
+count(*)
+1
+select count(*) from mysql.index_stats where table_name = 't1';
+count(*)
+1
+select count(*) from mysql.column_stats where table_name = 't1';
+count(*)
+1
+create or replace table t1 (b int primary key) as select 1 as b union all select 1;
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+select count(*) from mysql.table_stats where table_name = 't1';
+count(*)
+1
+select count(*) from mysql.index_stats where table_name = 't1';
+count(*)
+1
+select count(*) from mysql.column_stats where table_name = 't1';
+count(*)
+1
+create or replace table t1 (b int primary key);
+select count(*) from mysql.table_stats where table_name = 't1';
+count(*)
+0
+select count(*) from mysql.index_stats where table_name = 't1';
+count(*)
+0
+select count(*) from mysql.column_stats where table_name = 't1';
+count(*)
+0
+analyze table t1 persistent for all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	Table is already up to date
+set @@session.drop_before_create_or_replace='ON';
+create or replace table t1 (c int primary key) as select 1 as c union all select 1;
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+select count(*) from mysql.table_stats where table_name = 't1';
+count(*)
+0
+select count(*) from mysql.index_stats where table_name = 't1';
+count(*)
+0
+select count(*) from mysql.column_stats where table_name = 't1';
+count(*)
+0
+drop table if exists t1;
+Warnings:
+Note	1051	Unknown table 'test.t1'
+set @@session.drop_before_create_or_replace=default;
+#
 # End of 12.3 tests
 #

mysql-test/main/create_or_replace.test

@@ -908,6 +908,39 @@ CREATE OR REPLACE TABLE t1 (b INT) ENGINE=aria;
 SELECT 1;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-36501 EITS data is lost after failed attempt to
+--echo # CREATE OR REPLACE table
+--echo #
+create table t1 (a int primary key);
+insert into t1 values (1),(2);
+analyze table t1 persistent for all;
+select count(*) from mysql.table_stats where table_name = 't1';
+select count(*) from mysql.index_stats where table_name = 't1';
+select count(*) from mysql.column_stats where table_name = 't1';
+
+--error ER_DUP_ENTRY
+create or replace table t1 (b int primary key) as select 1 as b union all select 1;
+select count(*) from mysql.table_stats where table_name = 't1';
+select count(*) from mysql.index_stats where table_name = 't1';
+select count(*) from mysql.column_stats where table_name = 't1';
+
+create or replace table t1 (b int primary key);
+select count(*) from mysql.table_stats where table_name = 't1';
+select count(*) from mysql.index_stats where table_name = 't1';
+select count(*) from mysql.column_stats where table_name = 't1';
+analyze table t1 persistent for all;
+
+set @@session.drop_before_create_or_replace='ON';
+
+--error ER_DUP_ENTRY
+create or replace table t1 (c int primary key) as select 1 as c union all select 1;
+select count(*) from mysql.table_stats where table_name = 't1';
+select count(*) from mysql.index_stats where table_name = 't1';
+select count(*) from mysql.column_stats where table_name = 't1';
+drop table if exists t1;
+set @@session.drop_before_create_or_replace=default;
+
 --echo #
 --echo # End of 12.3 tests
 --echo #

sql/ddl_log.cc

@@ -1734,9 +1734,14 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
             break;
         }
       }
+      if (ddl_log_entry->extra_name.str)
+        delete_statistics_for_table(thd, &db, &ddl_log_entry->extra_name);
       break;
     case DDL_DROP_PHASE_RESET:
-      /* We have already logged all previous drop's. Clear the query */
+      /*
+        We only come here if DDL_DROP_PHASE_BINLOG fails.
+        We have already logged all previous drop's. Clear the query
+      */
       recovery_state.drop_table.length(recovery_state.drop_table_init_length);
       recovery_state.drop_view.length(recovery_state.drop_view_init_length);
       break;
@@ -3343,6 +3348,7 @@ bool ddl_log_drop_view_init(DDL_LOG_STATE *ddl_state,
    @param path          Table filepath without extension
    @param db            DB name
    @param table         Table name
+   @param orig_table    Original table name (in case of create or replace)
    @param flags         DDL_LOG_FLAG_FROM_IS_TMP or DDL_LOG_FLAG_TO_IS_TMP
 */
 
@@ -3353,6 +3359,7 @@ static bool ddl_log_drop(DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *path,
                          const LEX_CSTRING *db,
                          const LEX_CSTRING *table,
+                         const LEX_CSTRING *orig_table,
                          uint16 flags)
 {
   DDL_LOG_ENTRY ddl_log_entry;
@@ -3371,6 +3378,8 @@ static bool ddl_log_drop(DDL_LOG_STATE *ddl_state,
   ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(db);
   ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(table);
   ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
+  if (orig_table)
+    ddl_log_entry.extra_name=  *const_cast<LEX_CSTRING*>(orig_table);
   ddl_log_entry.phase=        (uchar) phase;
   ddl_log_entry.flags=        flags;
 
@@ -3402,6 +3411,7 @@ static bool ddl_log_drop(DDL_LOG_STATE *ddl_state,
    @param path          Table filepath without extension
    @param db            DB name
    @param table         Table name
+   @param orig_table    Original table name (in case of create or replace)
    @param flags         DDL_LOG_FLAG_FROM_IS_TMP or DDL_LOG_FLAG_TO_IS_TMP
 */
 
@@ -3410,12 +3420,13 @@ bool ddl_log_drop_table(DDL_LOG_STATE *ddl_state,
                         const LEX_CSTRING *path,
                         const LEX_CSTRING *db,
                         const LEX_CSTRING *table,
+                        const LEX_CSTRING *orig_table,
                         uint16 flags)
 {
   DBUG_ENTER("ddl_log_drop_table");
   DBUG_RETURN(ddl_log_drop(ddl_state,
                            DDL_LOG_DROP_TABLE_ACTION, DDL_DROP_PHASE_TABLE,
-                           hton, path, db, table, flags));
+                           hton, path, db, table, orig_table, flags));
 }
 
 
@@ -3434,7 +3445,8 @@ bool ddl_log_drop_view(DDL_LOG_STATE *ddl_state,
   DBUG_ENTER("ddl_log_drop_view");
   DBUG_RETURN(ddl_log_drop(ddl_state,
                            DDL_LOG_DROP_VIEW_ACTION, 0,
-                           (handlerton*) 0, path, db, table, 0));
+                           (handlerton*) 0, path, db, table, (LEX_CSTRING*) 0,
+                           0));
 }
 
 

sql/ddl_log.h

@@ -325,6 +325,7 @@ bool ddl_log_drop_table(DDL_LOG_STATE *ddl_state,
                         const LEX_CSTRING *path,
                         const LEX_CSTRING *db,
                         const LEX_CSTRING *table,
+                        const LEX_CSTRING *orig_table,
                         uint16 flags);
 bool ddl_log_drop_view(DDL_LOG_STATE *ddl_state,
                         const LEX_CSTRING *path,

sql/sql_table.cc

@@ -1351,9 +1351,9 @@ static uint32 get_comment(THD *thd, uint32 comment_pos,
   Used by create_or_replace().
 */
 
-int ddl_log_drop_tmp_table(THD *thd, DDL_LOG_STATE *ddl_log_state,
-                           TABLE_LIST *table,
-                           Table_specification_st *create_info)
+static int ddl_log_drop_tmp_table(THD *thd, DDL_LOG_STATE *ddl_log_state,
+                                  TABLE_LIST *table,
+                                  Table_specification_st *create_info)
 {
   bool res= 0;
   LEX_CSTRING comment= {"",0};
@@ -1377,6 +1377,9 @@ int ddl_log_drop_tmp_table(THD *thd, DDL_LOG_STATE *ddl_log_state,
   else
     res= ddl_log_drop_table(ddl_log_state, create_info->org_hton,
                             &cpath, db, &create_info->tmp_name,
+                            (!create_info->tmp_table() ?
+                             &create_info->alias /* original name */ :
+                             (LEX_CSTRING*) 0),
                             DDL_LOG_FLAG_FROM_IS_TMP);
   return res;
 }
@@ -1777,7 +1780,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
         res= ddl_log_drop_view(ddl_log_state, &cpath, &db, &table_name);
       else
         res= ddl_log_drop_table(ddl_log_state, hton, &cpath, &db, &table_name,
-                                0);
+                                (LEX_CSTRING*) 0, 0);
       if (res)
         goto err;
 
@@ -1865,7 +1868,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
       int ferror= 0;
       DBUG_ASSERT(!was_view);
 
-      if (ddl_log_drop_table(ddl_log_state, 0, &cpath, &db, &table_name, 0))
+      if (ddl_log_drop_table(ddl_log_state, 0, &cpath, &db, &table_name,
+                             (LEX_CSTRING *) 0, 0))
       {
         error= -1;
         goto err;
@@ -4724,6 +4728,7 @@ finalize_create_table(THD *thd, TABLE_LIST *orig_table, const char *query,
   }
   else if (table_was_deleted)
     ddl_log_update_xid(ddl_log_state_rm, thd->binlog_xid);
+
   ddl_log_update_xid(ddl_log_state_create, thd->binlog_xid);
 
   debug_crash_here("ddl_log_create_before_binlog");
@@ -5132,8 +5137,6 @@ int create_table_impl(THD *thd,
           goto err;
         }
 
-        (void) delete_statistics_for_table(thd, &db, &table_name);
-
         if (use_drop_table)
         {
           /*
@@ -5148,6 +5151,8 @@ int create_table_impl(THD *thd,
               save_unsafe_rollback_flags;
           }
 
+          (void) delete_statistics_for_table(thd, &db, &table_name);
+
           /* Remove normal table without logging. Keep tables locked */
           if (mysql_rm_table_no_locks(thd, &table_list, &thd->db,
                                       ddl_log_state_rm,
@@ -5178,6 +5183,7 @@ int create_table_impl(THD *thd,
           /* Rename the conflicting table to a temporary table name */
           bool force_if_exists= 0;
           rename_param param;
+          MDL_request mdl_request;
 
           create_info->org_hton= db_type;
           if (!(create_info->tmp_name.str= thd->alloc(64)))
@@ -5213,6 +5219,14 @@ int create_table_impl(THD *thd,
           DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db.str,
                                                      table_name.str,
                                                      MDL_EXCLUSIVE));
+          /*
+            Create an exclusive lock on the temporary table name.
+            Needed by InnoDB to not block purge.
+           */
+          MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
+                           db.str, create_info->tmp_name.str,
+                           MDL_EXCLUSIVE, MDL_TRANSACTION);
+          thd->mdl_context.acquire_lock(&mdl_request, 0);
 
           /*
             We have to reset partition_info from the CREATE TABLE as