MDEV-8075: DROP TEMPORARY TABLE not marked as ddl, causing optimistic parallel replication to fail

CREATE/DROP TEMPORARY TABLE are not safe to optimistically replicate in
parallel with other transactions, so they need to be marked as "ddl" in the
binlog.

This was already done for stand-alone CREATE/DROP TEMPORARY. But temporary
tables can also be created and dropped inside a BEGIN...END transaction, and
such transactions were not marked as ddl. Nor was the DROP TEMPORARY TABLE
statement emitted implicitly when a client connection is closed.

So this patch adds such ddl mark for the missing cases.

The difference to Kristian's original patch is mainly a fix in
mysql_trans_commit_alter_copy_data() to remember the unsafe_rollback_flags
over the temporary commit.

8 files changed, 74 insertions, 26 deletions

diff --git a/sql/handler.h b/sql/handler.h
index dad2b81b39c..1d4dded3971 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1470,16 +1470,30 @@ struct THD_TRANS
   static unsigned int const CREATED_TEMP_TABLE= 0x02;
   static unsigned int const DROPPED_TEMP_TABLE= 0x04;
   static unsigned int const DID_WAIT= 0x08;
+  static unsigned int const DID_DDL= 0x10;
 
   void mark_created_temp_table()
   {
     DBUG_PRINT("debug", ("mark_created_temp_table"));
     m_unsafe_rollback_flags|= CREATED_TEMP_TABLE;
   }
+  void mark_dropped_temp_table()
+  {
+    DBUG_PRINT("debug", ("mark_dropped_temp_table"));
+    m_unsafe_rollback_flags|= DROPPED_TEMP_TABLE;
+  }
+  bool has_created_dropped_temp_table() const {
+    return
+      (m_unsafe_rollback_flags & (CREATED_TEMP_TABLE|DROPPED_TEMP_TABLE)) != 0;
+  }
   void mark_trans_did_wait() { m_unsafe_rollback_flags|= DID_WAIT; }
   bool trans_did_wait() const {
     return (m_unsafe_rollback_flags & DID_WAIT) != 0;
   }
+  void mark_trans_did_ddl() { m_unsafe_rollback_flags|= DID_DDL; }
+  bool trans_did_ddl() const {
+    return (m_unsafe_rollback_flags & DID_DDL) != 0;
+  }
 
 };
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index f098664b1ba..4265a23df2b 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -6635,8 +6635,10 @@ Gtid_log_event::Gtid_log_event(THD *thd_arg, uint64 seq_no_arg,
   if (thd_arg->transaction.stmt.trans_did_wait() ||
       thd_arg->transaction.all.trans_did_wait())
     flags2|= FL_WAITED;
-  if (sql_command_flags[thd->lex->sql_command] &
-      (CF_DISALLOW_IN_RO_TRANS | CF_AUTO_COMMIT_TRANS))
+  if (thd_arg->transaction.stmt.trans_did_ddl() ||
+      thd_arg->transaction.stmt.has_created_dropped_temp_table() ||
+      thd_arg->transaction.all.trans_did_ddl() ||
+      thd_arg->transaction.all.has_created_dropped_temp_table())
     flags2|= FL_DDL;
   else if (is_transactional)
     flags2|= FL_TRANSACTIONAL;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 9c56b7d8f22..ae10c0b771d 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1245,6 +1245,7 @@ bool close_temporary_tables(THD *thd)
       thd->variables.character_set_client= cs_save;
 
       thd->get_stmt_da()->set_overwrite_status(true);
+      thd->transaction.stmt.mark_dropped_temp_table();
       if ((error= (mysql_bin_log.write(&qinfo) || error)))
       {
         /*
diff --git a/sql/sql_class.h b/sql/sql_class.h
index db1214ab26b..342902363c6 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -4048,6 +4048,16 @@ public:
   {
     main_lex.restore_set_statement_var();
   }
+  /* Copy relevant `stmt` transaction flags to `all` transaction. */
+  void merge_unsafe_rollback_flags()
+  {
+    if (transaction.stmt.modified_non_trans_table)
+      transaction.all.modified_non_trans_table= TRUE;
+    transaction.all.m_unsafe_rollback_flags|=
+      (transaction.stmt.m_unsafe_rollback_flags &
+       (THD_TRANS::DID_WAIT | THD_TRANS::CREATED_TEMP_TABLE |
+        THD_TRANS::DROPPED_TEMP_TABLE | THD_TRANS::DID_DDL));
+  }
 };
 
 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index e750e97194a..092b2154c61 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -4360,6 +4360,15 @@ void select_create::store_values(List<Item> &values)
 bool select_create::send_eof()
 {
   DBUG_ENTER("select_create::send_eof");
+
+  /*
+    The routine that writes the statement in the binary log
+    is in select_insert::prepare_eof(). For that reason, we
+    mark the flag at this point.
+  */
+  if (table->s->tmp_table)
+    thd->transaction.stmt.mark_created_temp_table();
+
   if (prepare_eof())
   {
     abort_result_set();
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 62d73f5306a..c4b32933b8a 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2847,6 +2847,7 @@ mysql_execute_command(THD *thd)
         goto error;
       }
     }
+    thd->transaction.stmt.mark_trans_did_ddl();
   }
 
 #ifndef DBUG_OFF
@@ -6936,8 +6937,7 @@ void THD::reset_for_next_command()
   if (!thd->in_multi_stmt_transaction_mode())
   {
     thd->variables.option_bits&= ~OPTION_KEEP_LOG;
-    thd->transaction.all.modified_non_trans_table= FALSE;
-    thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+    thd->transaction.all.reset();
   }
   DBUG_ASSERT(thd->security_ctx== &thd->main_security_ctx);
   thd->thread_specific_used= FALSE;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index db0d6d4f377..ace8ff1a7a9 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2560,6 +2560,9 @@ err:
   if (non_trans_tmp_table_deleted ||
       trans_tmp_table_deleted || non_tmp_table_deleted)
   {
+    if (non_trans_tmp_table_deleted || trans_tmp_table_deleted)
+      thd->transaction.stmt.mark_dropped_temp_table();
+
     query_cache_invalidate3(thd, tables, 0);
     if (!dont_log_query && mysql_bin_log.is_open())
     {
@@ -5073,6 +5076,9 @@ err:
   if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
     DBUG_RETURN(result);
 
+  if (create_info->tmp_table())
+    thd->transaction.stmt.mark_created_temp_table();
+
   /* Write log if no error or if we already deleted a table */
   if (!result || thd->log_current_statement)
   {
@@ -5567,13 +5573,17 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
     DBUG_PRINT("info",
                ("res: %d  tmp_table: %d  create_info->table: %p",
                 res, create_info->tmp_table(), local_create_info.table));
-    if (!res && create_info->tmp_table() && local_create_info.table)
+    if (create_info->tmp_table())
     {
-      /*
-        Remember that tmp table creation was logged so that we know if
-        we should log a delete of it.
-      */
-      local_create_info.table->s->table_creation_was_logged= 1;
+      thd->transaction.stmt.mark_created_temp_table();
+      if (!res && local_create_info.table)
+      {
+        /*
+          Remember that tmp table creation was logged so that we know if
+          we should log a delete of it.
+        */
+        local_create_info.table->s->table_creation_was_logged= 1;
+      }
     }
     do_logging= TRUE;
   }
@@ -9366,8 +9376,12 @@ bool mysql_trans_prepare_alter_copy_data(THD *thd)
 bool mysql_trans_commit_alter_copy_data(THD *thd)
 {
   bool error= FALSE;
+  uint save_unsafe_rollback_flags;
   DBUG_ENTER("mysql_trans_commit_alter_copy_data");
 
+  /* Save flags as transcommit_implicit_are_deleting_them */
+  save_unsafe_rollback_flags= thd->transaction.stmt.m_unsafe_rollback_flags;
+
   if (ha_enable_transaction(thd, TRUE))
     DBUG_RETURN(TRUE);
   
@@ -9382,6 +9396,7 @@ bool mysql_trans_commit_alter_copy_data(THD *thd)
   if (trans_commit_implicit(thd))
     error= TRUE;
 
+  thd->transaction.stmt.m_unsafe_rollback_flags= save_unsafe_rollback_flags;
   DBUG_RETURN(error);
 }
 
diff --git a/sql/transaction.cc b/sql/transaction.cc
index 8b188709ce6..f03cced7bc8 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -151,11 +151,10 @@ bool trans_begin(THD *thd, uint flags)
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
 
   /*
-    The following set should not be needed as the flag should always be 0
-    when we come here.  We should at some point change this to an assert.
+    The following set should not be needed as transaction state should
+    already be reset. We should at some point change this to an assert.
   */
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
   thd->has_waiter= false;
   thd->waiting_on_group_commit= false;
 
@@ -251,8 +250,7 @@ bool trans_commit(THD *thd)
   else
     (void) RUN_HOOK(transaction, after_commit, (thd, FALSE));
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
   thd->lex->start_transaction_opt= 0;
 
   DBUG_RETURN(MY_TEST(res));
@@ -299,8 +297,7 @@ bool trans_commit_implicit(THD *thd)
   }
 
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
 
   /*
     Upon implicit commit, reset the current transaction
@@ -345,8 +342,7 @@ bool trans_rollback(THD *thd)
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
   /* Reset the binlog transaction marker */
   thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
   thd->lex->start_transaction_opt= 0;
 
   DBUG_RETURN(MY_TEST(res));
@@ -390,8 +386,7 @@ bool trans_rollback_implicit(THD *thd)
     preserve backward compatibility.
   */
   thd->variables.option_bits&= ~(OPTION_KEEP_LOG);
-  thd->transaction.all.modified_non_trans_table= false;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
 
   /* Rollback should clear transaction_rollback_request flag. */
   DBUG_ASSERT(! thd->transaction_rollback_request);
@@ -427,6 +422,8 @@ bool trans_commit_stmt(THD *thd)
   */
   DBUG_ASSERT(! thd->in_sub_stmt);
 
+  thd->merge_unsafe_rollback_flags();
+
   if (thd->transaction.stmt.ha_list)
   {
     if (WSREP_ON)
@@ -481,6 +478,8 @@ bool trans_rollback_stmt(THD *thd)
   */
   DBUG_ASSERT(! thd->in_sub_stmt);
 
+  thd->merge_unsafe_rollback_flags();
+
   if (thd->transaction.stmt.ha_list)
   {
     if (WSREP_ON)
@@ -904,8 +903,7 @@ bool trans_xa_commit(THD *thd)
   }
 
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
   thd->server_status&=
     ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
   DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));
@@ -960,8 +958,7 @@ bool trans_xa_rollback(THD *thd)
   res= xa_trans_force_rollback(thd);
 
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->transaction.all.modified_non_trans_table= FALSE;
-  thd->transaction.all.m_unsafe_rollback_flags&= ~THD_TRANS::DID_WAIT;
+  thd->transaction.all.reset();
   thd->server_status&=
     ~(SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY);
   DBUG_PRINT("info", ("clearing SERVER_STATUS_IN_TRANS"));