BUG#51894 Replication failure with SBR on DROP TEMPORARY TABLE inside a

          transaction
BUG#52616 Temp table prevents switch binlog format from STATEMENT to ROW

Before the WL#2687 and BUG#46364, every non-transactional change that happened
after a transactional change was written to trx-cache and flushed upon
committing the transaction. WL#2687 and BUG#46364 changed this behavior and
non-transactional changes are now written to the binary log upon committing
the statement.

A binary log event is identified as transactional or non-transactional through
a flag in the Log_event which is set taking into account the underlie storage
engine on what it is stems from. In the current bug, this flag was not being
set properly when the DROP TEMPORARY TABLE was executed.

However, while fixing this bug we figured out that changes to temporary tables
should be always written to the trx-cache if there is an on-going transaction.
Otherwise, binlog events in the reversed order would be produced.

Regarding concurrency, keeping changes to temporary tables in the trx-cache is
also safe as temporary tables are only visible to the owner connection.

In this patch, we classify the following statements as unsafe:
   1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp

   2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam

   3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam

On the other hand, the following statements are classified as safe:

   1 - INSERT INTO t_innodb SELECT * FROM t_myisam_temp

   2 - INSERT INTO t_myisam_temp SELECT * FROM t_innodb

The patch also guarantees that transactions that have a DROP TEMPORARY are
always written to the binary log regardless of the mode and the outcome:
commit or rollback. In particular, the DROP TEMPORARY is extended with the
IF EXISTS clause when the current statement logging format is set to row.

Finally, the patch allows to switch from STATEMENT to MIXED/ROW when there
are temporary tables but the contrary is not possible.

mysql-test/extra/rpl_tests/rpl_binlog_max_cache_size.test:
  Updated the test case because 
     CREATE TEMPORARY TABLE t_innodb_temp SELECT * FROM t_myisam is not unsafe.
mysql-test/extra/rpl_tests/rpl_implicit_commit_binlog.test:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/extra/rpl_tests/rpl_innodb.test:
  Removed comments from the test case that became false after the patch.
mysql-test/extra/rpl_tests/rpl_loaddata.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
mysql-test/include/ctype_utf8_table.inc:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
mysql-test/r/ctype_cp932_binlog_stm.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/r/binlog_database.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/r/binlog_innodb_row.result:
  Updated the result file.
mysql-test/suite/binlog/r/binlog_multi_engine.result:
  Updated the unsafe message.
mysql-test/suite/binlog/r/binlog_row_binlog.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/r/binlog_row_drop_tmp_tbl.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/r/binlog_row_mix_innodb_myisam.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/r/binlog_stm_binlog.result:
  Updated the result file.
mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/binlog/t/binlog_tmp_table.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
mysql-test/suite/ndb/r/ndb_binlog_format.result:
  Updated the unsafe message.
mysql-test/suite/rpl/r/rpl_concurrency_error.result:
  Updated the unsafe message.
mysql-test/suite/rpl/r/rpl_mixed_binlog_max_cache_size.result:
  Updated the result file because 
     CREATE TEMPORARY TABLE t_innodb_temp SELECT * FROM t_myisam is not unsafe.
mysql-test/suite/rpl/r/rpl_mixed_implicit_commit_binlog.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_mixed_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_mixed_row_innodb.result:
  Added some comments to ease the understanding of the result file.
mysql-test/suite/rpl/r/rpl_non_direct_mixed_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_non_direct_row_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_non_direct_stm_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_row_drop.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_row_implicit_commit_binlog.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_row_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_stm_binlog_max_cache_size.result:
  Updated the result file because 
     CREATE TEMPORARY TABLE t_innodb_temp SELECT * FROM t_myisam is not unsafe.
mysql-test/suite/rpl/r/rpl_stm_implicit_commit_binlog.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_stm_innodb.result:
  Added some comments to ease the understanding of the result file.
mysql-test/suite/rpl/r/rpl_stm_mixing_engines.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl/r/rpl_stm_stop_middle_group.result:
  Updated the unsafe message.
mysql-test/suite/rpl/r/rpl_temp_temporary.result:
  Added a test case.
mysql-test/suite/rpl/t/rpl000013.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
mysql-test/suite/rpl/t/rpl_misc_functions.test:
  Suppressed warning messages.
mysql-test/suite/rpl/t/rpl_temp_table.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
mysql-test/suite/rpl/t/rpl_temp_temporary.test:
  Added a test case.
mysql-test/suite/rpl/t/rpl_temporary.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
mysql-test/suite/rpl_ndb/r/rpl_ndb_row_implicit_commit_binlog.result:
  Updated the test case due to the new rules: changes to
  temporary tables are written to the binary log in the
  boundaries of a transaction if there is any.
mysql-test/suite/rpl_ndb/r/rpl_truncate_7ndb.result:
  Updated the test case to remove references to positions
  in the binary log.
mysql-test/suite/rpl_ndb/t/rpl_truncate_7ndb.test:
  Updated the test case to remove references to positions
  in the binary log.
mysql-test/t/create_select_tmp.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
mysql-test/t/ctype_cp932_binlog_stm.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
mysql-test/t/mysqlbinlog.test:
  Suppressed warning messages due to the following cases:
  
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
     3 - CREATE TEMPORARY TABLE t_myisam_temp SELECT * FROM t_myisam
sql/log.cc:
  Improved the code by creating several functions to hide decision
  on type of engine changed, commit/abort, etc:  
  
  . stmt_has_updated_non_trans_table
  
  . trans_has_updated_non_trans_table
  
  . ending_trans
  
  Updated the binlog_rollback function and the use of the 
  OPTION_KEEP_LOG which indincates when a temporary table was
  either created or dropped and as such the command must be 
  logged if not in MIXED mode and even while rolling back the
  transaction.
sql/log.h:
  Improved the code by creating several functions to hide decision
  on type of engine changed, commit/abort, etc.
sql/log_event.cc:
  Removed the setting of the OPTION_KEEP_LOG as it is related to CREATE
  TEMPORARY and DROP TEMPORARY and not to the type of engine (i.e.
  transactional or non-transactional).
sql/log_event_old.cc:
  Removed the setting of the OPTION_KEEP_LOG as it is related to CREATE
  TEMPORARY and DROP TEMPORARY and not to the type of engine (i.e.
  transactional or non-transactional).
sql/share/errmsg-utf8.txt:
  Updated the unsafe message.
sql/sql_class.cc:
  Classifies the following statements as unsafe:
     1 - INSERT INTO t_myisam SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_myisam
  
  On the other hand, the following statements are classified as safe:
  
     1 - INSERT INTO t_innodb SELECT * FROM t_myisam_temp
  
     2 - INSERT INTO t_myisam_temp SELECT * FROM t_innodb
sql/sql_class.h:
  It allows to switch from STATEMENT to MIXED/ROW when there are temporary
  tables but the contrary is not possible.
sql/sql_table.cc:
  Fixed the case that a DROP/DROP TEMPORARY that affects a temporary table in MIXED
  mode is written as a DROP TEMPORARY TABLE IF EXISTS because the table may not exist in the slave and due to the IF EXISTS token an error will never happen
  while processing the statement in the slave.
  
  Removed a function that was not being used.

8 files changed, 173 insertions, 98 deletions

diff --git a/sql/log.cc b/sql/log.cc
index 92ffcc58961..18f812d96bf 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1576,7 +1576,7 @@ binlog_truncate_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr, bool all)
     inside a transaction, we reset the transaction cache.
   */
   thd->binlog_remove_pending_rows_event(TRUE, is_transactional);
-  if (all || !thd->in_multi_stmt_transaction())
+  if (ending_trans(thd, all))
   {
     if (cache_mngr->trx_cache.has_incident())
       error= mysql_bin_log.write_incident(thd, TRUE);
@@ -1682,12 +1682,11 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   DBUG_ENTER("binlog_commit");
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
-  bool const in_transaction= thd->in_multi_stmt_transaction();
 
   DBUG_PRINT("debug",
              ("all: %d, in_transaction: %s, all.modified_non_trans_table: %s, stmt.modified_non_trans_table: %s",
               all,
-              YESNO(in_transaction),
+              YESNO(thd->in_multi_stmt_transaction()),
               YESNO(thd->transaction.all.modified_non_trans_table),
               YESNO(thd->transaction.stmt.modified_non_trans_table)));
 
@@ -1711,7 +1710,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
      - We are in a transaction and a full transaction is committed.
     Otherwise, we accumulate the changes.
   */
-  if (!in_transaction || all)
+  if (ending_trans(thd, all))
   {
     Query_log_event qev(thd, STRING_WITH_LEN("COMMIT"), TRUE, FALSE, TRUE, 0);
     error= binlog_flush_trx_cache(thd, cache_mngr, &qev);
@@ -1789,29 +1788,28 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   {  
     /*
       We flush the cache wrapped in a beging/rollback if:
-        . aborting a transcation that modified a non-transactional table or;
-        . aborting a statement that modified both transactional and
-         non-transctional tables but which is not in the boundaries of any
-         transaction;
+        . aborting a single or multi-statement transaction and;
+        . the format is STMT and non-trans engines were updated or;
         . the OPTION_KEEP_LOG is activate.
     */
-    if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
-        ((all && thd->transaction.all.modified_non_trans_table) ||
-        (!all && thd->transaction.stmt.modified_non_trans_table &&
-        !thd->in_multi_stmt_transaction()) ||
-        (thd->variables.option_bits & OPTION_KEEP_LOG)))
+    if (ending_trans(thd, all) &&
+        ((thd->variables.option_bits & OPTION_KEEP_LOG) ||
+         (trans_has_updated_non_trans_table(thd) &&
+          thd->variables.binlog_format == BINLOG_FORMAT_STMT)))
     {
       Query_log_event qev(thd, STRING_WITH_LEN("ROLLBACK"), TRUE, FALSE, TRUE, 0);
       error= binlog_flush_trx_cache(thd, cache_mngr, &qev);
     }
     /*
-      Otherwise, we simply truncate the cache as there is no change on
-      non-transactional tables as follows.
+      Truncate the cache if:
+        . aborting a single or multi-statement transaction or;
+        . the OPTION_KEEP_LOG is not activate and;
+        . the format is not STMT or no non-trans were updated.
     */
-    else if (all || (!all &&
-                     (!thd->transaction.stmt.modified_non_trans_table ||
-                      !thd->in_multi_stmt_transaction() || 
-                      thd->variables.binlog_format != BINLOG_FORMAT_STMT)))
+    else if (ending_trans(thd, all) ||
+             (!(thd->variables.option_bits & OPTION_KEEP_LOG) &&
+              ((!stmt_has_updated_non_trans_table(thd) ||
+               thd->variables.binlog_format != BINLOG_FORMAT_STMT))))
       error= binlog_truncate_trx_cache(thd, cache_mngr, all);
   }
 
@@ -1909,7 +1907,7 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
     non-transactional table. Otherwise, truncate the binlog cache starting
     from the SAVEPOINT command.
   */
-  if (unlikely(thd->transaction.all.modified_non_trans_table ||
+  if (unlikely(trans_has_updated_non_trans_table(thd) ||
                (thd->variables.option_bits & OPTION_KEEP_LOG)))
   {
     int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
@@ -4257,6 +4255,49 @@ bool use_trans_cache(const THD* thd, bool is_transactional)
      (is_transactional || !cache_mngr->trx_cache.empty()));
 }
 
+/**
+  This function checks if a transaction, either a multi-statement
+  or a single statement transaction is about to commit or not.
+
+  @param thd The client thread that executed the current statement.
+  @param all Committing a transaction (i.e. TRUE) or a statement
+             (i.e. FALSE).
+  @return
+    @c true if committing a transaction, otherwise @c false.
+*/
+bool ending_trans(THD* thd, const bool all)
+{
+  return (all || (!all && !thd->in_multi_stmt_transaction()));
+}
+
+/**
+  This function checks if a non-transactional table was updated by
+  the current transaction.
+
+  @param thd The client thread that executed the current statement.
+  @return
+    @c true if a non-transactional table was updated, @c false
+    otherwise.
+*/
+bool trans_has_updated_non_trans_table(const THD* thd)
+{
+  return (thd->transaction.all.modified_non_trans_table ||
+          thd->transaction.stmt.modified_non_trans_table);
+}
+
+/**
+  This function checks if a non-transactional table was updated by the
+  current statement.
+
+  @param thd The client thread that executed the current statement.
+  @return
+    @c true if a non-transactional table was updated, @c false otherwise.
+*/
+bool stmt_has_updated_non_trans_table(const THD* thd)
+{
+  return (thd->transaction.stmt.modified_non_trans_table);
+}
+
 /*
   These functions are placed in this file since they need access to
   binlog_hton, which has internal linkage.
@@ -4539,7 +4580,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
     {
       set_write_error(thd);
       if (check_write_error(thd) && cache_data &&
-          thd->transaction.stmt.modified_non_trans_table)
+          stmt_has_updated_non_trans_table(thd))
         cache_data->set_incident();
       DBUG_RETURN(1);
     }
@@ -4740,7 +4781,7 @@ unlock:
     {
       set_write_error(thd);
       if (check_write_error(thd) && cache_data &&
-        thd->transaction.stmt.modified_non_trans_table)
+          stmt_has_updated_non_trans_table(thd))
         cache_data->set_incident();
     }
   }
diff --git a/sql/log.h b/sql/log.h
index 71c27f50f78..12e02969485 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -26,6 +26,9 @@ class Format_description_log_event;
 bool trans_has_updated_trans_table(const THD* thd);
 bool stmt_has_updated_trans_table(const THD *thd);
 bool use_trans_cache(const THD* thd, bool is_transactional);
+bool ending_trans(THD* thd, const bool all);
+bool trans_has_updated_non_trans_table(const THD* thd);
+bool stmt_has_updated_non_trans_table(const THD* thd);
 
 /*
   Transaction Coordinator log - a base abstract class
diff --git a/sql/log_event.cc b/sql/log_event.cc
index fc097414726..3a52b72909a 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -679,6 +679,7 @@ Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
   server_id=	thd->server_id;
   when=		thd->start_time;
   cache_type= (using_trans || stmt_has_updated_trans_table(thd)
+               || thd->thread_specific_used
                ? Log_event::EVENT_TRANSACTIONAL_CACHE :
                Log_event::EVENT_STMT_CACHE);
 }
@@ -2470,6 +2471,7 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   */
   LEX *lex= thd->lex;
   bool implicit_commit= FALSE;
+  bool force_trans= FALSE;
   cache_type= Log_event::EVENT_INVALID_CACHE;
   switch (lex->sql_command)
   {
@@ -2483,14 +2485,16 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
       implicit_commit= TRUE;
       break;
     case SQLCOM_DROP_TABLE:
-      implicit_commit= !(lex->drop_temporary && thd->in_multi_stmt_transaction());
+      force_trans= lex->drop_temporary && thd->in_multi_stmt_transaction();
+      implicit_commit= !force_trans;
       break;
     case SQLCOM_ALTER_TABLE:
     case SQLCOM_CREATE_TABLE:
-      implicit_commit= !((lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
-                   thd->in_multi_stmt_transaction()) &&
-                  !(lex->select_lex.item_list.elements &&
-                   thd->is_current_stmt_binlog_format_row());
+      force_trans= (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
+                    thd->in_multi_stmt_transaction();
+      implicit_commit= !force_trans &&
+                       !(lex->select_lex.item_list.elements &&
+                         thd->is_current_stmt_binlog_format_row());
       break;
     case SQLCOM_SET_OPTION:
       implicit_commit= (lex->autocommit ? TRUE : FALSE);
@@ -2548,7 +2552,8 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   }
   else
   {
-    cache_type= (using_trans || stmt_has_updated_trans_table(thd)
+    cache_type= ((using_trans || stmt_has_updated_trans_table(thd) ||
+                 force_trans || thd->thread_specific_used)
                  ? Log_event::EVENT_TRANSACTIONAL_CACHE :
                  Log_event::EVENT_STMT_CACHE);
   }
@@ -7733,12 +7738,6 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
       clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
       error= 0;
     }
-
-    if (!use_trans_cache())
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      thd->variables.option_bits|= OPTION_KEEP_LOG;
-    }
   } // if (table)
 
   
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index a80e8f4b1a2..f6c5b5d1023 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -241,11 +241,6 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
     DBUG_EXECUTE_IF("stop_slave_middle_group",
                     const_cast<Relay_log_info*>(rli)->abort_slave= 1;);
     error= do_after_row_operations(table, error);
-    if (!ev->use_trans_cache())
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      ev_thd->variables.option_bits|= OPTION_KEEP_LOG;
-    }
   }
 
   if (error)
@@ -1683,11 +1678,6 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
     DBUG_EXECUTE_IF("stop_slave_middle_group",
                     const_cast<Relay_log_info*>(rli)->abort_slave= 1;);
     error= do_after_row_operations(rli, error);
-    if (!use_trans_cache())
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      thd->variables.option_bits|= OPTION_KEEP_LOG;
-    }
   } // if (table)
 
   if (error)
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 927de5adff4..1308ad0ebcf 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6335,4 +6335,4 @@ ER_BINLOG_UNSAFE_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE
   eng "Mixing self-logging and non-self-logging engines in a statement is unsafe."
 
 ER_BINLOG_UNSAFE_MIXED_STATEMENT
-  eng "Statements that read from both transactional and non-transactional tables and write to any of them are unsafe."
+  eng "Statements that read from both transactional (or a temporary table of any engine type) and non-transactional tables and write to any of them are unsafe."
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d07bfdc6abf..88f67175d02 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3661,19 +3661,80 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         if (prev_write_table && prev_write_table->file->ht !=
             table->table->file->ht)
           multi_write_engine= TRUE;
+        /*
+          Every temporary table must be always written to the binary
+          log in transaction boundaries and as such we artificially
+          classify them as transactional.
+
+          Indirectly, this avoids classifying a temporary table created
+          on a non-transactional engine as unsafe when it is modified
+          after any transactional table:
+
+          BEGIN;
+            INSERT INTO innodb_t VALUES (1);
+            INSERT INTO myisam_t_temp VALUES (1);
+          COMMIT;
+
+          BINARY LOG:
+
+          BEGIN;
+            INSERT INTO innodb_t VALUES (1);
+            INSERT INTO myisam_t_temp VALUES (1);
+          COMMIT;
+        */
         all_trans_write_engines= all_trans_write_engines &&
-                                 table->table->file->has_transactions();
+                                 (table->table->file->has_transactions() ||
+                                  table->table->s->tmp_table);
         prev_write_table= table->table;
         flags_write_all_set &= flags;
         flags_write_some_set |= flags;
       }
       flags_some_set |= flags;
-      if (prev_access_table && prev_access_table->file->ht != table->table->file->ht)
+      /*
+        The mixture of non-transactional and transactional tables must
+        identified and classified as unsafe. However, a temporary table
+        must be always handled as a transactional table. Based on that,
+        we have the following statements classified as mixed and by
+        consequence as unsafe:
+
+        1: INSERT INTO myisam_t SELECT * FROM innodb_t;
+
+        2: INSERT INTO innodb_t SELECT * FROM myisam_t;
+
+        3: INSERT INTO myisam_t SELECT * FROM myisam_t_temp;
+
+        4: INSERT INTO myisam_t_temp SELECT * FROM myisam_t;
+
+        5: CREATE TEMPORARY TABLE myisam_t_temp SELECT * FROM mysiam_t;
+
+        The following statements are not considered mixed and as such
+        are safe:
+
+        1: INSERT INTO innodb_t SELECT * FROM myisam_t_temp;
+
+        2: INSERT INTO myisam_t_temp SELECT * FROM innodb_t_temp;
+      */
+      if (!trans_non_trans_access_engines && prev_access_table &&
+          (lex->sql_command != SQLCOM_CREATE_TABLE ||
+          (lex->sql_command == SQLCOM_CREATE_TABLE &&
+          (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))))
       {
+        my_bool prev_trans;
+        my_bool act_trans;
+        if (prev_access_table->s->tmp_table || table->table->s->tmp_table)
+        {
+          prev_trans= prev_access_table->s->tmp_table ? TRUE :
+                     prev_access_table->file->has_transactions();
+          act_trans= table->table->s->tmp_table ? TRUE :
+                    table->table->file->has_transactions();
+        }
+        else
+        {
+          prev_trans= prev_access_table->file->has_transactions();
+          act_trans= table->table->file->has_transactions();
+        }
+        trans_non_trans_access_engines= (prev_trans != act_trans);
         multi_access_engine= TRUE;
-        trans_non_trans_access_engines= trans_non_trans_access_engines ||
-          (prev_access_table->file->has_transactions() !=
-           table->table->file->has_transactions());
       }
       prev_access_table= table->table;
     }
@@ -3710,6 +3771,12 @@ int THD::decide_logging_format(TABLE_LIST *tables)
 
       2: INSERT INTO innodb_t SELECT * FROM myisam_t;
 
+      3: INSERT INTO myisam_t SELECT * FROM myisam_t_temp;
+
+      4: INSERT INTO myisam_t_temp SELECT * FROM myisam_t;
+
+      5: CREATE TEMPORARY TABLE myisam_t_temp SELECT * FROM mysiam_t;
+
       are classified as unsafe to ensure that in mixed mode the execution is
       completely safe and equivalent to the row mode. Consider the following
       statements and sessions (connections) to understand the reason:
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 950db01d1cd..c75d91baada 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2531,11 +2531,11 @@ public:
                ("temporary_tables: %s, in_sub_stmt: %s, system_thread: %s",
                 YESNO(temporary_tables), YESNO(in_sub_stmt),
                 show_system_thread(system_thread)));
-    if ((temporary_tables == NULL) && (in_sub_stmt == 0))
+    if (in_sub_stmt == 0)
     {
       if (variables.binlog_format == BINLOG_FORMAT_ROW)
         set_current_stmt_binlog_format_row();
-      else
+      else if (temporary_tables == NULL)
         clear_current_stmt_binlog_format_row();
     }
     DBUG_VOID_RETURN;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c3a39354ff1..709fee07aac 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1998,8 +1998,13 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     case  0:
       // removed temporary table
       tmp_table_deleted= 1;
-      if (thd->variables.binlog_format == BINLOG_FORMAT_MIXED &&
-          thd->is_current_stmt_binlog_format_row())
+      /*
+        One needs to always log any temporary table drop if the current
+        statement logging format is set to row. This happens because one
+        might have created a temporary table while the statement logging
+        format was statement and then switched to mixed or row format.
+      */
+      if (thd->is_current_stmt_binlog_format_row())
       {
         if (built_tmp_query.is_empty()) 
         {
@@ -2191,7 +2196,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
           deleted one or more non-temporary tables (and no temporary
           tables).  In this case, we can write the original query into
           the binary log.
-         */
+        */
         error |= write_bin_log(thd, !error, thd->query(), thd->query_length());
       }
       else if (thd->is_current_stmt_binlog_format_row() &&
@@ -2218,11 +2223,12 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
         }
 
         /*
-          One needs to always log any temporary table drop, if:
-            1. thread logging format is mixed mode; AND
-            2. current statement logging format is set to row.
+          One needs to always log any temporary table drop if the current
+          statement logging format is set to row. This happens because one
+          might have created a temporary table while the statement logging
+          format was statement and then switched to mixed or row format.
         */
-        if (thd->variables.binlog_format == BINLOG_FORMAT_MIXED)
+        if (thd->is_current_stmt_binlog_format_row())
         {
           /*
             In this case we have deleted some temporary tables but we are using
@@ -2233,8 +2239,14 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
           */
           built_tmp_query.chop();                  // Chop of the last comma
           built_tmp_query.append(" /* generated by server */");
-          error|= write_bin_log(thd, !error, built_tmp_query.ptr(), built_tmp_query.length(),
-                                thd->in_multi_stmt_transaction());
+          /*
+            We cannot call the write_bin_log as we do not care about any errors
+            in the master as the statement is always DROP TEMPORARY TABLE IF EXISTS
+            and as such there will be no errors in the slave.
+          */
+          error|= thd->binlog_query(THD::STMT_QUERY_TYPE, built_tmp_query.ptr(),
+                                    built_tmp_query.length(), FALSE, FALSE, FALSE,
+                                    0);
         }
       }
 
@@ -3749,43 +3761,6 @@ void sp_prepare_create_field(THD *thd, Create_field *sql_field)
   (void) prepare_blob_field(thd, sql_field);
 }
 
-
-/*
-  Write CREATE TABLE binlog
-
-  SYNOPSIS
-    write_create_table_bin_log()
-    thd               Thread object
-    create_info       Create information
-    internal_tmp_table  Set to 1 if this is an internal temporary table
-
-  DESCRIPTION
-    This function only is called in mysql_create_table_no_lock and
-    mysql_create_table
-
-  RETURN VALUES
-    NONE
- */
-static inline int write_create_table_bin_log(THD *thd,
-                                             const HA_CREATE_INFO *create_info,
-                                             bool internal_tmp_table)
-{
-  /*
-    Don't write statement if:
-    - It is an internal temporary table,
-    - Row-based logging is used and it we are creating a temporary table, or
-    - The binary log is not open.
-    Otherwise, the statement shall be binlogged.
-   */
-  if (!internal_tmp_table &&
-      (!thd->is_current_stmt_binlog_format_row() ||
-       (thd->is_current_stmt_binlog_format_row() &&
-        !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
-    return write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-  return 0;
-}
-
-
 /*
   Create a table