BUG#53560 CREATE TEMP./DROP TEMP. are not binglogged correctly after a failed statement

This patch fixes two problems described as follows:

1 - If there is an on-going transaction and a temporary table is created or
dropped, any failed statement that follows the "create" or "drop commands"
triggers a rollback and by consequence the slave will go out sync because
the binary log will have a wrong sequence of events.

To fix the problem, we changed the expression that evaluates when the
cache should be flushed after either the rollback of a statment or
transaction.

2 - When a "CREATE TEMPORARY TABLE SELECT * FROM" was executed the
OPTION_KEEP_LOG was not set into the thd->options. For that reason, if
the transaction had updated only transactional engines and was rolled
back at the end (.e.g due to a deadlock) the changes were not written
to the binary log, including the creation of the temporary table.
      
To fix the problem, we have set the OPTION_KEEP_LOG into the thd->options
when a "CREATE TEMPORARY TABLE SELECT * FROM" is executed.

8 files changed, 190 insertions, 41 deletions

diff --git a/mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result b/mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result
index b3f9baed2dd..a26fcc1dc1a 100644
--- a/mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result
+++ b/mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result
@@ -364,6 +364,9 @@ master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 values (10,10)
 master-bin.000001	#	Query	#	#	BEGIN
 master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t2 values (100,100)
 master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t2 values (101,101)
+master-bin.000001	#	Query	#	#	ROLLBACK
 master-bin.000001	#	Query	#	#	use `test`; DROP TABLE t1,t2
 reset master;
 create table t1 (a int) engine=innodb;
diff --git a/mysql-test/suite/rpl/r/rpl_temporary_errors.result b/mysql-test/suite/rpl/r/rpl_temporary_errors.result
index d14380a6369..2bd1b691901 100644
--- a/mysql-test/suite/rpl/r/rpl_temporary_errors.result
+++ b/mysql-test/suite/rpl/r/rpl_temporary_errors.result
@@ -81,4 +81,61 @@ Last_SQL_Errno	0
 Last_SQL_Error	
 DROP TABLE t1;
 **** On Master ****
+SET SQL_LOG_BIN= 0;
 DROP TABLE t1;
+SET SQL_LOG_BIN= 1;
+SET SESSION BINLOG_FORMAT=MIXED;
+CREATE TABLE t_myisam (id INT, PRIMARY KEY (id)) engine= MyIsam;
+INSERT INTO t_myisam (id) VALUES(1);
+CREATE TABLE t_innodb (id INT) engine= Innodb;
+INSERT INTO t_innodb (id) VALUES(1);
+BEGIN;
+INSERT INTO t_innodb(id) VALUES(2);
+INSERT INTO t_myisam(id) VALUES(3);
+CREATE TEMPORARY TABLE x (id INT);
+INSERT INTO t_myisam(id) VALUES(4),(1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+INSERT INTO t_innodb(id) VALUES(5);
+COMMIT;
+SELECT * FROM t_innodb;
+id
+1
+2
+5
+SELECT * FROM t_myisam;
+id
+1
+3
+4
+SELECT * FROM t_innodb;
+id
+1
+2
+5
+SELECT * FROM t_myisam;
+id
+1
+3
+4
+BEGIN;
+CREATE TEMPORARY TABLE tmp2 SELECT * FROM t_innodb;
+INSERT INTO t_innodb(id) VALUES(1);
+INSERT INTO t_innodb(id) VALUES(1);
+ROLLBACK;
+Warnings:
+Warning	1196	Some non-transactional changed tables couldn't be rolled back
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_innodb(id) VALUES(2)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_myisam(id) VALUES(3)
+master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE x (id INT)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_myisam(id) VALUES(4),(1)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_innodb(id) VALUES(5)
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; CREATE TEMPORARY TABLE tmp2 SELECT * FROM t_innodb
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_innodb(id) VALUES(1)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t_innodb(id) VALUES(1)
+master-bin.000001	#	Query	#	#	ROLLBACK
+DROP TABLE t_myisam, t_innodb;
diff --git a/mysql-test/suite/rpl/t/rpl_temporary_errors.test b/mysql-test/suite/rpl/t/rpl_temporary_errors.test
index 3b373e00a62..4bc374cdca7 100644
--- a/mysql-test/suite/rpl/t/rpl_temporary_errors.test
+++ b/mysql-test/suite/rpl/t/rpl_temporary_errors.test
@@ -1,4 +1,5 @@
 source include/master-slave.inc;
+source include/have_innodb.inc;
 
 call mtr.add_suppression("Deadlock found");
 
@@ -30,4 +31,43 @@ DROP TABLE t1;
 
 --echo **** On Master ****
 connection master;
+SET SQL_LOG_BIN= 0;
 DROP TABLE t1;
+SET SQL_LOG_BIN= 1;
+
+# BUG#Bug #53259 Unsafe statement binlogged in statement format w/MyIsam temp tables
+#
+SET SESSION BINLOG_FORMAT=MIXED;
+CREATE TABLE t_myisam (id INT, PRIMARY KEY (id)) engine= MyIsam;
+INSERT INTO t_myisam (id) VALUES(1);
+CREATE TABLE t_innodb (id INT) engine= Innodb;
+INSERT INTO t_innodb (id) VALUES(1);
+
+let $binlog_start= query_get_value("SHOW MASTER STATUS", Position, 1);
+BEGIN;
+INSERT INTO t_innodb(id) VALUES(2);
+INSERT INTO t_myisam(id) VALUES(3);
+CREATE TEMPORARY TABLE x (id INT);
+--error 1062
+INSERT INTO t_myisam(id) VALUES(4),(1);
+INSERT INTO t_innodb(id) VALUES(5);
+COMMIT;
+
+SELECT * FROM t_innodb;
+SELECT * FROM t_myisam;
+
+--sync_slave_with_master
+
+SELECT * FROM t_innodb;
+SELECT * FROM t_myisam;
+
+--connection master
+
+BEGIN;
+CREATE TEMPORARY TABLE tmp2 SELECT * FROM t_innodb;
+INSERT INTO t_innodb(id) VALUES(1);
+INSERT INTO t_innodb(id) VALUES(1);
+ROLLBACK;
+source include/show_binlog_events.inc;
+
+DROP TABLE t_myisam, t_innodb;
diff --git a/sql/log.cc b/sql/log.cc
index 7d820b48c43..79a6b4d6cb0 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1510,27 +1510,23 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   }
 
   /*
-    We commit the transaction if:
+    We flush the cache if:
 
-     - We are not in a transaction and committing a statement, or
-
-     - We are in a transaction and a full transaction is committed
+     - we are committing a transaction;
+     - and no statement was committed before and just non-transactional
+       tables were updated.
 
     Otherwise, we accumulate the statement
   */
-  ulonglong const in_transaction=
-    thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN);
   DBUG_PRINT("debug",
-             ("all: %d, empty: %s, in_transaction: %s, all.modified_non_trans_table: %s, stmt.modified_non_trans_table: %s",
+             ("all: %d, empty: %s, all.modified_non_trans_table: %s, stmt.modified_non_trans_table: %s",
               all,
               YESNO(trx_data->empty()),
-              YESNO(in_transaction),
               YESNO(thd->transaction.all.modified_non_trans_table),
               YESNO(thd->transaction.stmt.modified_non_trans_table)));
-  if (!in_transaction || all ||
-      (!all && !trx_data->at_least_one_stmt_committed &&
-      !stmt_has_updated_trans_table(thd) &&
-      thd->transaction.stmt.modified_non_trans_table))
+  if (ending_trans(thd, all) ||
+      (trans_has_no_stmt_committed(thd, all) &&
+       !stmt_has_updated_trans_table(thd) && stmt_has_updated_non_trans_table(thd)))
   {
     Query_log_event qev(thd, STRING_WITH_LEN("COMMIT"), TRUE, TRUE, 0);
     error= binlog_end_trans(thd, trx_data, &qev, all);
@@ -1593,7 +1589,7 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
       On the other hand, if a statement is transactional, we just safely roll it
       back.
     */
-    if ((thd->transaction.stmt.modified_non_trans_table ||
+    if ((stmt_has_updated_non_trans_table(thd) ||
         (thd->options & OPTION_KEEP_LOG)) &&
         mysql_bin_log.check_write_error(thd))
       trx_data->set_incident();
@@ -1603,19 +1599,18 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   {
    /*
       We flush the cache with a rollback, wrapped in a beging/rollback if:
-        . aborting a transaction that modified a non-transactional table;
+        . aborting a transaction that modified a non-transactional table or
+          the OPTION_KEEP_LOG is activate.
         . aborting a statement that modified both transactional and
           non-transactional tables but which is not in the boundaries of any
           transaction or there was no early change;
-        . the OPTION_KEEP_LOG is activate.
     */
-    if ((all && thd->transaction.all.modified_non_trans_table) ||
-        (!all && thd->transaction.stmt.modified_non_trans_table &&
-         !(thd->options & (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT))) ||
-        (!all && thd->transaction.stmt.modified_non_trans_table &&
-         !trx_data->at_least_one_stmt_committed &&
-         thd->current_stmt_binlog_row_based) ||
-        ((thd->options & OPTION_KEEP_LOG)))
+    if ((ending_trans(thd, all) &&
+        (trans_has_updated_non_trans_table(thd) ||
+         (thd->options & OPTION_KEEP_LOG))) ||
+        (trans_has_no_stmt_committed(thd, all) &&
+         stmt_has_updated_non_trans_table(thd) &&
+         thd->current_stmt_binlog_row_based))
     {
       Query_log_event qev(thd, STRING_WITH_LEN("ROLLBACK"), TRUE, TRUE, 0);
       error= binlog_end_trans(thd, trx_data, &qev, all);
@@ -1624,8 +1619,8 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
       Otherwise, we simply truncate the cache as there is no change on
       non-transactional tables as follows.
     */
-    else if ((all && !thd->transaction.all.modified_non_trans_table) ||
-          (!all && !thd->transaction.stmt.modified_non_trans_table))
+    else if (ending_trans(thd, all) ||
+             (!(thd->options & OPTION_KEEP_LOG) && !stmt_has_updated_non_trans_table(thd)))
       error= binlog_end_trans(thd, trx_data, 0, all);
   }
   if (!all)
@@ -1721,7 +1716,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->options & OPTION_KEEP_LOG)))
   {
     String log_query;
@@ -3934,6 +3929,67 @@ bool MYSQL_BIN_LOG::is_query_in_union(THD *thd, query_id_t query_id_param)
           query_id_param >= thd->binlog_evt_union.first_query_id);
 }
 
+/**
+  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(const THD* thd, const bool all)
+{
+  return (all || (!all && !(thd->options & 
+                  (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT))));
+}
+
+/**
+  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 any statement was committed and cached.
+
+  @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 at a statement was committed and cached, @c false
+    otherwise.
+*/
+bool trans_has_no_stmt_committed(const THD* thd, bool all)
+{
+  binlog_trx_data *const trx_data=
+    (binlog_trx_data*) thd_get_ha_data(thd, binlog_hton);
+
+  return (!all && !trx_data->at_least_one_stmt_committed);
+}
+
+/**
+  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
diff --git a/sql/log.h b/sql/log.h
index 5af51e14d80..8d3880d9171 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -20,6 +20,11 @@ class Relay_log_info;
 
 class Format_description_log_event;
 
+bool ending_trans(const THD* thd, const bool all);
+bool trans_has_updated_non_trans_table(const THD* thd);
+bool trans_has_no_stmt_committed(const THD* thd, const bool all);
+bool stmt_has_updated_non_trans_table(const THD* thd);
+
 /*
   Transaction Coordinator log - a base abstract class
   for two different implementations
diff --git a/sql/log_event.cc b/sql/log_event.cc
index a8e227fa99b..617d1188984 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -7544,12 +7544,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 (!cache_stmt)
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      thd->options|= OPTION_KEEP_LOG;
-    }
   } // if (table)
 
   /*
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index df162761b35..be389acafe8 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -229,11 +229,6 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
     DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event",
                     const_cast<Relay_log_info*>(rli)->abort_slave= 1;);
     error= do_after_row_operations(table, error);
-    if (!ev->cache_stmt)
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      ev_thd->options|= OPTION_KEEP_LOG;
-    }
   }
 
   /*
@@ -1755,11 +1750,6 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
     DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event",
                     const_cast<Relay_log_info*>(rli)->abort_slave= 1;);
     error= do_after_row_operations(rli, error);
-    if (!cache_stmt)
-    {
-      DBUG_PRINT("info", ("Marked that we need to keep log"));
-      thd->options|= OPTION_KEEP_LOG;
-    }
   } // if (table)
 
   /*
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 93d80164ffb..98e7f187fc4 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2713,6 +2713,10 @@ mysql_execute_command(THD *thd)
           }
         }
 
+        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
+        if (create_info.options & HA_LEX_CREATE_TMP_TABLE)
+          thd->options|= OPTION_KEEP_LOG;
+
         /*
           select_create is currently not re-execution friendly and
           needs to be created for every execution of a PS/SP.