BUG#22864 (Rollback following CREATE... SELECT discards 'CREATE TABLE'

from log):
When row-based logging is used, the CREATE-SELECT is written as two
parts: as a CREATE TABLE statement and as the rows for the table. For
both transactional and non-transactional tables, the CREATE TABLE
statement was written to the transaction cache, as were the rows, and
on statement end, the entire transaction cache was written to the binary
log if the table was non-transactional. For transactional tables, the
events were kept in the transaction cache until end of transaction (or
statement that were not part of a transaction).

For the case when AUTOCOMMIT=0 and we are creating a transactional table
using a create select, we would then keep the CREATE TABLE statement and
the rows for the CREATE-SELECT, while executing the following statements.
On a rollback, the transaction cache would then be cleared, which would
also remove the CREATE TABLE statement. Hence no table would be created
on the slave, while there is an empty table on the master.

This relates to BUG#22865 where the table being created exists on the
master, but not on the slave during insertion of rows into the newly
created table. This occurs since the CREATE TABLE statement were still
in the transaction cache until the statement finished executing, and
possibly longer if the table was transactional.

This patch changes the behaviour of the CREATE-SELECT statement by
adding an implicit commit at the end of the statement when creating
non-temporary tables. Hence, non-temporary tables will be written to the
binary log on completion, and in the even of AUTOCOMMIT=0, a new
transaction will be started. Temporary tables do not commit an ongoing
transaction: neither as a pre- not a post-commit.

The events for both transactional and non-transactional tables are
saved in the transaction cache, and written to the binary log at end
of the statement.


mysql-test/r/rpl_row_create_table.result:
  Result change
mysql-test/t/rpl_row_create_table.test:
  Requring InnoDB for slave as well.
  Adding test CREATE-SELECT that is rolled back explicitly.
  Changing binlog positions.
sql/log.cc:
  Adding helper class to handle lock/unlock of mutexes using RAII.
  Factoring out code into write_cache() function to transaction cache
    to binary log.
  Adding function THD::binlog_flush_transaction_cache() to flush the
    transaction cache to the binary log file.
  Factoring out code into binlog_set_stmt_begin() to set the beginning
    of statement savepoint.
  Clearing before statement point when transaction cache is truncated
   so that these points are out of range.
sql/log.h:
  Adding method MYSQL_BIN_LOG::write_cache()
sql/log_event.h:
  Replicating OPTION_NOT_AUTOCOMMIT flag (see changeset comment)
sql/mysql_priv.h:
  Although left-shifting signed integer values is well-defined,
  it has potential for strange errors. Using unsigned long long
  instead of signed long long since this is the type of the options
  flags.
sql/slave.cc:
  Adding printout of transaction-critical thread flags.
sql/sql_class.h:
  Adding function THD::binlog_flush_transaction_cache()
  Adding function THD::binlog_set_stmt_begin()
sql/sql_insert.cc:
  Adding code to cache events for a CREATE-SELECT statement.
  Disabling binlog for SBR (but not RBR) when sending error for select part
  of CREATE-SELECT statement.
  Adding implicit commit at end of statement for non-temporary tables.
mysql-test/t/rpl_row_create_table-slave.opt:
  New BitKeeper file ``mysql-test/t/rpl_row_create_table-slave.opt''

10 files changed, 564 insertions, 93 deletions

diff --git a/mysql-test/r/rpl_row_create_table.result b/mysql-test/r/rpl_row_create_table.result
index 03388f59b8c..8f587fb5796 100644
--- a/mysql-test/r/rpl_row_create_table.result
+++ b/mysql-test/r/rpl_row_create_table.result
@@ -127,7 +127,7 @@ NULL	5	10
 NULL	6	12
 CREATE TABLE t7 (UNIQUE(b)) SELECT a,b FROM tt3;
 ERROR 23000: Duplicate entry '2' for key 'b'
-SHOW BINLOG EVENTS FROM 1256;
+SHOW BINLOG EVENTS FROM 1118;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 CREATE TABLE t7 (a INT, b INT UNIQUE);
 INSERT INTO t7 SELECT a,b FROM tt3;
@@ -212,3 +212,192 @@ Create Table	CREATE TABLE `t9` (
   `a` int(11) DEFAULT NULL,
   `b` int(11) DEFAULT NULL
 ) ENGINE=MEMORY DEFAULT CHARSET=latin1
+DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8,t9;
+STOP SLAVE;
+SET GLOBAL storage_engine=@storage_engine;
+START SLAVE;
+================ BUG#22864 ================
+STOP SLAVE;
+RESET SLAVE;
+RESET MASTER;
+START SLAVE;
+SET AUTOCOMMIT=0;
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1),(2),(3);
+CREATE TABLE t2 ENGINE=INNODB SELECT * FROM t1;
+ROLLBACK;
+CREATE TABLE t3 ENGINE=INNODB SELECT * FROM t1;
+INSERT INTO t3 VALUES (4),(5),(6);
+ROLLBACK;
+CREATE TABLE t4 ENGINE=INNODB SELECT * FROM t1;
+INSERT INTO t1 VALUES (4),(5),(6);
+ROLLBACK;
+Warnings:
+Warning	1196	Some non-transactional changed tables couldn't be rolled back
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+t3
+t4
+SELECT   TABLE_NAME,ENGINE
+FROM   INFORMATION_SCHEMA.TABLES
+WHERE   TABLE_NAME LIKE 't_'
+ORDER BY TABLE_NAME;
+TABLE_NAME	ENGINE
+t1	MyISAM
+t2	InnoDB
+t3	InnoDB
+t4	InnoDB
+SELECT * FROM t1 ORDER BY a;
+a
+1
+2
+3
+4
+5
+6
+SELECT * FROM t2 ORDER BY a;
+a
+1
+2
+3
+SELECT * FROM t3 ORDER BY a;
+a
+1
+2
+3
+SELECT * FROM t4 ORDER BY a;
+a
+1
+2
+3
+SHOW BINLOG EVENTS;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	4	Format_desc	1	102	Server ver: #, Binlog ver: #
+master-bin.000001	102	Query	1	188	use `test`; CREATE TABLE t1 (a INT)
+master-bin.000001	188	Table_map	1	227	table_id: # (test.t1)
+master-bin.000001	227	Write_rows	1	271	table_id: # flags: STMT_END_F
+master-bin.000001	271	Query	1	339	use `test`; BEGIN
+master-bin.000001	339	Query	1	125	use `test`; CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB
+master-bin.000001	464	Table_map	1	164	table_id: # (test.t2)
+master-bin.000001	503	Write_rows	1	208	table_id: # flags: STMT_END_F
+master-bin.000001	547	Xid	1	574	COMMIT /* XID */
+master-bin.000001	574	Query	1	642	use `test`; BEGIN
+master-bin.000001	642	Query	1	125	use `test`; CREATE TABLE `t3` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB
+master-bin.000001	767	Table_map	1	164	table_id: # (test.t3)
+master-bin.000001	806	Write_rows	1	208	table_id: # flags: STMT_END_F
+master-bin.000001	850	Xid	1	877	COMMIT /* XID */
+master-bin.000001	877	Query	1	945	use `test`; BEGIN
+master-bin.000001	945	Query	1	125	use `test`; CREATE TABLE `t4` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB
+master-bin.000001	1070	Table_map	1	164	table_id: # (test.t4)
+master-bin.000001	1109	Write_rows	1	208	table_id: # flags: STMT_END_F
+master-bin.000001	1153	Xid	1	1180	COMMIT /* XID */
+master-bin.000001	1180	Table_map	1	1219	table_id: # (test.t1)
+master-bin.000001	1219	Write_rows	1	1263	table_id: # flags: STMT_END_F
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+t3
+t4
+SELECT   TABLE_NAME,ENGINE
+FROM   INFORMATION_SCHEMA.TABLES
+WHERE   TABLE_NAME LIKE 't_'
+ORDER BY TABLE_NAME;
+TABLE_NAME	ENGINE
+t1	MyISAM
+t2	InnoDB
+t3	InnoDB
+t4	InnoDB
+SELECT * FROM t1 ORDER BY a;
+a
+1
+2
+3
+4
+5
+6
+SELECT * FROM t2 ORDER BY a;
+a
+1
+2
+3
+SELECT * FROM t3 ORDER BY a;
+a
+1
+2
+3
+SELECT * FROM t4 ORDER BY a;
+a
+1
+2
+3
+DROP TABLE IF EXISTS t1,t2,t3,t4;
+SET AUTOCOMMIT=1;
+STOP SLAVE;
+RESET SLAVE;
+RESET MASTER;
+START SLAVE;
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1),(2),(3);
+CREATE TABLE t2 (a INT) ENGINE=INNODB;
+BEGIN;
+INSERT INTO t2 SELECT a*a FROM t1;
+CREATE TEMPORARY TABLE tt1
+SELECT a+1 AS a
+FROM t1
+WHERE a MOD 2 = 1;
+INSERT INTO t2 SELECT a+2 FROM tt1;
+COMMIT;
+SELECT * FROM t2 ORDER BY a;
+a
+1
+4
+4
+6
+9
+SHOW BINLOG EVENTS;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	4	Format_desc	1	102	Server ver: #, Binlog ver: #
+master-bin.000001	102	Query	1	188	use `test`; CREATE TABLE t1 (a INT)
+master-bin.000001	188	Table_map	1	227	table_id: # (test.t1)
+master-bin.000001	227	Write_rows	1	271	table_id: # flags: STMT_END_F
+master-bin.000001	271	Query	1	371	use `test`; CREATE TABLE t2 (a INT) ENGINE=INNODB
+master-bin.000001	371	Query	1	439	use `test`; BEGIN
+master-bin.000001	439	Table_map	1	39	table_id: # (test.t2)
+master-bin.000001	478	Write_rows	1	83	table_id: # flags: STMT_END_F
+master-bin.000001	522	Table_map	1	122	table_id: # (test.t2)
+master-bin.000001	561	Write_rows	1	161	table_id: # flags: STMT_END_F
+master-bin.000001	600	Xid	1	627	COMMIT /* XID */
+SELECT * FROM t2 ORDER BY a;
+a
+1
+4
+4
+6
+9
+TRUNCATE TABLE t2;
+BEGIN;
+INSERT INTO t2 SELECT a*a FROM t1;
+CREATE TEMPORARY TABLE tt2
+SELECT a+1 AS a
+FROM t1
+WHERE a MOD 2 = 1;
+INSERT INTO t2 SELECT a+2 FROM tt2;
+ROLLBACK;
+SELECT * FROM t2 ORDER BY a;
+a
+SHOW BINLOG EVENTS FROM 627;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	627	Query	1	80	use `test`; TRUNCATE TABLE t2
+master-bin.000001	707	Xid	1	734	COMMIT /* XID */
+SELECT * FROM t2 ORDER BY a;
+a
+DROP TABLE t1,t2;
diff --git a/mysql-test/t/rpl_row_create_table-slave.opt b/mysql-test/t/rpl_row_create_table-slave.opt
new file mode 100644
index 00000000000..627becdbfb5
--- /dev/null
+++ b/mysql-test/t/rpl_row_create_table-slave.opt
@@ -0,0 +1 @@
+--innodb
diff --git a/mysql-test/t/rpl_row_create_table.test b/mysql-test/t/rpl_row_create_table.test
index 3a711e5b496..6accf0c8bef 100644
--- a/mysql-test/t/rpl_row_create_table.test
+++ b/mysql-test/t/rpl_row_create_table.test
@@ -2,6 +2,10 @@
 
 --source include/have_binlog_format_row.inc
 --source include/master-slave.inc
+--source include/have_innodb.inc
+connection slave;
+--source include/have_innodb.inc
+connection master;
 
 # Bug#18326: Do not lock table for writing during prepare of statement
 # The use of the ps protocol causes extra table maps in the binlog, so
@@ -31,7 +35,7 @@ CREATE TABLE t2 (a INT, b INT) ENGINE=Merge;
 CREATE TABLE t3 (a INT, b INT) CHARSET=utf8;
 CREATE TABLE t4 (a INT, b INT) ENGINE=Merge CHARSET=utf8;
 --replace_column 1 # 4 # 5 #
---replace_regex /table_id: [0-9]+/table_id: #/
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
 --query_vertical SHOW BINLOG EVENTS FROM 212
 --echo **** On Master ****
 --query_vertical SHOW CREATE TABLE t1
@@ -66,8 +70,8 @@ connection master;
 --error 1062
 CREATE TABLE t7 (UNIQUE(b)) SELECT a,b FROM tt3;
 # Shouldn't be written to the binary log
---replace_regex /table_id: [0-9]+/table_id: #/
-SHOW BINLOG EVENTS FROM 1256;
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
+SHOW BINLOG EVENTS FROM 1118;
 
 # Test that INSERT-SELECT works the same way as for SBR.
 CREATE TABLE t7 (a INT, b INT UNIQUE);
@@ -75,7 +79,7 @@ CREATE TABLE t7 (a INT, b INT UNIQUE);
 INSERT INTO t7 SELECT a,b FROM tt3;
 SELECT * FROM t7 ORDER BY a,b;
 # Should be written to the binary log
---replace_regex /table_id: [0-9]+/table_id: #/
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
 SHOW BINLOG EVENTS FROM 1118;
 sync_slave_with_master;
 SELECT * FROM t7 ORDER BY a,b;
@@ -86,7 +90,7 @@ INSERT INTO tt4 VALUES (4,8), (5,10), (6,12);
 BEGIN;
 INSERT INTO t7 SELECT a,b FROM tt4;
 ROLLBACK;
---replace_regex /table_id: [0-9]+/table_id: #/
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
 SHOW BINLOG EVENTS FROM 1314;
 SELECT * FROM t7 ORDER BY a,b;
 sync_slave_with_master;
@@ -101,7 +105,7 @@ CREATE TEMPORARY TABLE tt7 SELECT 1;
 --echo **** On Master ****
 --query_vertical SHOW CREATE TABLE t8
 --query_vertical SHOW CREATE TABLE t9
---replace_regex /table_id: [0-9]+/table_id: #/
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
 SHOW BINLOG EVENTS FROM 1410;
 sync_slave_with_master;
 --echo **** On Slave ****
@@ -109,12 +113,117 @@ sync_slave_with_master;
 --query_vertical SHOW CREATE TABLE t9
 
 connection master;
---disable_query_log
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8,t9;
 sync_slave_with_master;
 # Here we reset the value of the default storage engine
 STOP SLAVE;
 SET GLOBAL storage_engine=@storage_engine;
 START SLAVE;
---enable_query_log
 --enable_ps_protocol
+
+# BUG#22864 (Rollback following CREATE ... SELECT discards 'CREATE
+# table' from log):
+--echo ================ BUG#22864 ================
+connection slave;
+STOP SLAVE;
+RESET SLAVE;
+connection master;
+RESET MASTER;
+connection slave;
+START SLAVE;
+connection master;
+SET AUTOCOMMIT=0;
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1),(2),(3);
+
+CREATE TABLE t2 ENGINE=INNODB SELECT * FROM t1;
+ROLLBACK;
+
+CREATE TABLE t3 ENGINE=INNODB SELECT * FROM t1;
+INSERT INTO t3 VALUES (4),(5),(6);
+ROLLBACK;
+
+CREATE TABLE t4 ENGINE=INNODB SELECT * FROM t1;
+INSERT INTO t1 VALUES (4),(5),(6);
+ROLLBACK;
+
+SHOW TABLES;
+SELECT   TABLE_NAME,ENGINE
+  FROM   INFORMATION_SCHEMA.TABLES
+ WHERE   TABLE_NAME LIKE 't_'
+ORDER BY TABLE_NAME;
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+SELECT * FROM t3 ORDER BY a;
+SELECT * FROM t4 ORDER BY a;
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /Server ver: .*, Binlog ver: .*/Server ver: #, Binlog ver: #/ /table_id: [0-9]+/table_id: #/
+SHOW BINLOG EVENTS;
+sync_slave_with_master;
+SHOW TABLES;
+SELECT   TABLE_NAME,ENGINE
+  FROM   INFORMATION_SCHEMA.TABLES
+ WHERE   TABLE_NAME LIKE 't_'
+ORDER BY TABLE_NAME;
+SELECT * FROM t1 ORDER BY a;
+SELECT * FROM t2 ORDER BY a;
+SELECT * FROM t3 ORDER BY a;
+SELECT * FROM t4 ORDER BY a;
+
+connection master;
+DROP TABLE IF EXISTS t1,t2,t3,t4;
+SET AUTOCOMMIT=1;
+sync_slave_with_master;
+
+# Some tests with temporary tables
+connection slave;
+STOP SLAVE;
+RESET SLAVE;
+
+connection master;
+RESET MASTER;
+
+connection slave;
+START SLAVE;
+
+connection master;
+CREATE TABLE t1 (a INT);
+INSERT INTO t1 VALUES (1),(2),(3);
+
+CREATE TABLE t2 (a INT) ENGINE=INNODB;
+
+BEGIN;
+INSERT INTO t2 SELECT a*a FROM t1;
+CREATE TEMPORARY TABLE tt1
+SELECT a+1 AS a
+  FROM t1
+ WHERE a MOD 2 = 1;
+INSERT INTO t2 SELECT a+2 FROM tt1;
+COMMIT;
+
+SELECT * FROM t2 ORDER BY a;
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /Server ver: .*, Binlog ver: .*/Server ver: #, Binlog ver: #/ /table_id: [0-9]+/table_id: #/
+SHOW BINLOG EVENTS;
+sync_slave_with_master;
+SELECT * FROM t2 ORDER BY a;
+
+connection master;
+TRUNCATE TABLE t2;
+
+BEGIN;
+INSERT INTO t2 SELECT a*a FROM t1;
+CREATE TEMPORARY TABLE tt2
+SELECT a+1 AS a
+  FROM t1
+ WHERE a MOD 2 = 1;
+INSERT INTO t2 SELECT a+2 FROM tt2;
+ROLLBACK;
+
+SELECT * FROM t2 ORDER BY a;
+--replace_regex /\/\* xid=.* \*\//\/* XID *\// /Server ver: .*, Binlog ver: .*/Server ver: #, Binlog ver: #/ /table_id: [0-9]+/table_id: #/
+SHOW BINLOG EVENTS FROM 627;
+sync_slave_with_master;
+SELECT * FROM t2 ORDER BY a;
+
+connection master;
+DROP TABLE t1,t2;
+sync_slave_with_master;
diff --git a/sql/log.cc b/sql/log.cc
index a1ed9bd6df3..1bb1b9a33e6 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -82,6 +82,41 @@ char *make_default_log_name(char *buff,const char* log_ext)
 }
 
 /*
+  Helper class to hold a mutex for the duration of the
+  block.
+
+  Eliminates the need for explicit unlocking of mutexes on, e.g.,
+  error returns.  On passing a null pointer, the sentry will not do
+  anything.
+ */
+class Mutex_sentry
+{
+public:
+  Mutex_sentry(pthread_mutex_t *mutex)
+    : m_mutex(mutex)
+  {
+    if (m_mutex)
+      pthread_mutex_lock(mutex);
+  }
+
+  ~Mutex_sentry()
+  {
+    if (m_mutex)
+      pthread_mutex_unlock(m_mutex);
+#ifndef DBUG_OFF
+    m_mutex= 0;
+#endif
+  }
+
+private:
+  pthread_mutex_t *m_mutex;
+
+  // It's not allowed to copy this object in any way
+  Mutex_sentry(Mutex_sentry const&);
+  void operator=(Mutex_sentry const&);
+};
+
+/*
   Helper class to store binary log transaction data.
 */
 class binlog_trx_data {
@@ -121,11 +156,17 @@ public:
    */
   void truncate(my_off_t pos)
   {
+    DBUG_PRINT("info", ("truncating to position %lu", pos));
+    DBUG_PRINT("info", ("before_stmt_pos=%lu", pos));
 #ifdef HAVE_ROW_BASED_REPLICATION
     delete pending();
     set_pending(0);
 #endif
     reinit_io_cache(&trans_log, WRITE_CACHE, pos, 0, 0);
+#ifdef HAVE_ROW_BASED_REPLICATION
+    if (pos < before_stmt_pos)
+      before_stmt_pos= MY_OFF_T_UNDEF;
+#endif
   }
 
   /*
@@ -1416,12 +1457,11 @@ binlog_end_trans(THD *thd, binlog_trx_data *trx_data,
 
       If rolling back a statement in a transaction, we truncate the
       transaction cache to remove the statement.
-
      */
     if (all || !(thd->options & (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT)))
       trx_data->reset();
-    else
-      trx_data->truncate(trx_data->before_stmt_pos); // ...statement
+    else                                        // ...statement
+      trx_data->truncate(trx_data->before_stmt_pos);
 
     /*
       We need to step the table map version on a rollback to ensure
@@ -2010,7 +2050,7 @@ bool MYSQL_QUERY_LOG::write(time_t event_time, const char *user_host,
           goto err;
 
     /* command_type, thread_id */
-    length= my_snprintf(buff, 32, "%5ld ", thread_id);
+    length= my_snprintf(buff, 32, "%5ld ", static_cast<long>(thread_id));
 
     if (my_b_write(&log_file, (byte*) buff, length))
       goto err;
@@ -3338,18 +3378,7 @@ THD::binlog_start_trans_and_stmt()
   if (trx_data == NULL ||
       trx_data->before_stmt_pos == MY_OFF_T_UNDEF)
   {
-    /*
-      The call to binlog_trans_log_savepos() might create the trx_data
-      structure, if it didn't exist before, so we save the position
-      into an auto variable and then write it into the transaction
-      data for the binary log (i.e., trx_data).
-    */
-    my_off_t pos= 0;
-    binlog_trans_log_savepos(this, &pos);
-    trx_data= (binlog_trx_data*) ha_data[binlog_hton->slot];
-
-    trx_data->before_stmt_pos= pos;
-
+    this->binlog_set_stmt_begin();
     if (options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
       trans_register_ha(this, TRUE, binlog_hton);
     trans_register_ha(this, FALSE, binlog_hton);
@@ -3357,6 +3386,51 @@ THD::binlog_start_trans_and_stmt()
   DBUG_VOID_RETURN;
 }
 
+void THD::binlog_set_stmt_begin() {
+  binlog_trx_data *trx_data=
+    (binlog_trx_data*) ha_data[binlog_hton->slot];
+
+  /*
+    The call to binlog_trans_log_savepos() might create the trx_data
+    structure, if it didn't exist before, so we save the position
+    into an auto variable and then write it into the transaction
+    data for the binary log (i.e., trx_data).
+  */
+  my_off_t pos= 0;
+  binlog_trans_log_savepos(this, &pos);
+  trx_data= (binlog_trx_data*) ha_data[binlog_hton->slot];
+  trx_data->before_stmt_pos= pos;
+}
+
+int THD::binlog_flush_transaction_cache()
+{
+  DBUG_ENTER("binlog_flush_transaction_cache");
+  binlog_trx_data *trx_data= (binlog_trx_data*) ha_data[binlog_hton->slot];
+  DBUG_PRINT("enter", ("trx_data=0x%lu", trx_data));
+  if (trx_data)
+    DBUG_PRINT("enter", ("trx_data->before_stmt_pos=%u",
+                         trx_data->before_stmt_pos));
+
+  /*
+    Write the transaction cache to the binary log.  We don't flush and
+    sync the log file since we don't know if more will be written to
+    it. If the caller want the log file sync:ed, the caller has to do
+    it.
+
+    The transaction data is only reset upon a successful write of the
+    cache to the binary log.
+  */
+
+  if (trx_data && likely(mysql_bin_log.is_open())) {
+    if (int error= mysql_bin_log.write_cache(&trx_data->trans_log, true, true))
+      DBUG_RETURN(error);
+    trx_data->reset();
+  }
+
+  DBUG_RETURN(0);
+}
+
+
 /*
   Write a table map to the binary log.
  */
@@ -3768,12 +3842,48 @@ uint MYSQL_BIN_LOG::next_file_id()
 
 
 /*
+  Write the contents of a cache to the binary log.
+
+  SYNOPSIS
+    write_cache()
+    cache    Cache to write to the binary log
+    lock_log True if the LOCK_log mutex should be aquired, false otherwise
+    sync_log True if the log should be flushed and sync:ed
+
+  DESCRIPTION
+    Write the contents of the cache to the binary log. The cache will
+    be reset as a READ_CACHE to be able to read the contents from it.
+ */
+
+int MYSQL_BIN_LOG::write_cache(IO_CACHE *cache, bool lock_log, bool sync_log)
+{
+  Mutex_sentry sentry(lock_log ? &LOCK_log : NULL);
+
+  if (reinit_io_cache(cache, READ_CACHE, 0, 0, 0))
+    return ER_ERROR_ON_WRITE;
+  uint bytes= my_b_bytes_in_cache(cache);
+  do
+  {
+    if (my_b_write(&log_file, cache->read_pos, bytes))
+      return ER_ERROR_ON_WRITE;
+    cache->read_pos= cache->read_end;
+  } while ((bytes= my_b_fill(cache)));
+
+  if (sync_log)
+    flush_and_sync();
+
+  return 0;                                     // All OK
+}
+
+/*
   Write a cached log entry to the binary log
 
   SYNOPSIS
     write()
     thd
     cache		The cache to copy to the binlog
+    commit_event        The commit event to print after writing the
+                        contents of the cache.
 
   NOTE
     - We only come here if there is something in the cache.
@@ -3833,20 +3943,10 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event)
         if (qinfo.write(&log_file))
           goto err;
       }
-      /* Read from the file used to cache the queries .*/
-      if (reinit_io_cache(cache, READ_CACHE, 0, 0, 0))
-        goto err;
-      length=my_b_bytes_in_cache(cache);
-      DBUG_EXECUTE_IF("half_binlogged_transaction", length-=100;);
-      do
-      {
-        /* Write data to the binary log file */
-        if (my_b_write(&log_file, cache->read_pos, length))
-          goto err;
-        cache->read_pos=cache->read_end;		// Mark buffer used up
-        DBUG_EXECUTE_IF("half_binlogged_transaction", goto DBUG_skip_commit;);
-      } while ((length=my_b_fill(cache)));
 
+      if ((write_error= write_cache(cache, false, false)))
+        goto err;
+      
       if (commit_event && commit_event->write(&log_file))
         goto err;
 #ifndef DBUG_OFF
diff --git a/sql/log.h b/sql/log.h
index 8f75601f02b..d82065ce0ff 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -338,6 +338,8 @@ public:
   bool write(Log_event* event_info); // binary log write
   bool write(THD *thd, IO_CACHE *cache, Log_event *commit_event);
 
+  int  write_cache(IO_CACHE *cache, bool lock_log, bool flush_and_sync);
+
   void start_union_events(THD *thd);
   void stop_union_events(THD *thd);
   bool is_query_in_union(THD *thd, query_id_t query_id_param);
diff --git a/sql/log_event.h b/sql/log_event.h
index 81ce2f18b4d..41e0f8aed46 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -405,12 +405,18 @@ struct sql_ex_info
    either, as the manual says (because a too big in-memory temp table is
    automatically written to disk).
 */
-#define OPTIONS_WRITTEN_TO_BIN_LOG (OPTION_AUTO_IS_NULL | \
-OPTION_NO_FOREIGN_KEY_CHECKS | OPTION_RELAXED_UNIQUE_CHECKS)
+#define OPTIONS_WRITTEN_TO_BIN_LOG \
+  (OPTION_AUTO_IS_NULL | OPTION_NO_FOREIGN_KEY_CHECKS |  \
+   OPTION_RELAXED_UNIQUE_CHECKS | OPTION_NOT_AUTOCOMMIT)
 
-#if OPTIONS_WRITTEN_TO_BIN_LOG != ((1L << 14) | (1L << 26) | (1L << 27))
+/* Shouldn't be defined before */
+#define EXPECTED_OPTIONS \
+  ((ULL(1) << 14) | (ULL(1) << 26) | (ULL(1) << 27) | (ULL(1) << 19))
+
+#if OPTIONS_WRITTEN_TO_BIN_LOG != EXPECTED_OPTIONS
 #error OPTIONS_WRITTEN_TO_BIN_LOG must NOT change their values!
 #endif
+#undef EXPECTED_OPTIONS         /* You shouldn't use this one */
 
 enum Log_event_type
 {
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 339ca9d965a..5e0d24a5c1f 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -296,54 +296,54 @@ MY_LOCALE *my_locale_by_name(const char *name);
    TODO: separate three contexts above, move them to separate bitfields.
 */
 
-#define SELECT_DISTINCT         (LL(1) << 0)       // SELECT, user
-#define SELECT_STRAIGHT_JOIN    (LL(1) << 1)       // SELECT, user
-#define SELECT_DESCRIBE         (LL(1) << 2)       // SELECT, user
-#define SELECT_SMALL_RESULT     (LL(1) << 3)       // SELECT, user
-#define SELECT_BIG_RESULT       (LL(1) << 4)       // SELECT, user
-#define OPTION_FOUND_ROWS       (LL(1) << 5)       // SELECT, user
-#define OPTION_TO_QUERY_CACHE   (LL(1) << 6)       // SELECT, user
-#define SELECT_NO_JOIN_CACHE    (LL(1) << 7)       // intern
-#define OPTION_BIG_TABLES       (LL(1) << 8)       // THD, user
-#define OPTION_BIG_SELECTS      (LL(1) << 9)       // THD, user
-#define OPTION_LOG_OFF          (LL(1) << 10)      // THD, user
-#define OPTION_QUOTE_SHOW_CREATE (LL(1) << 11)     // THD, user
-#define TMP_TABLE_ALL_COLUMNS   (LL(1) << 12)      // SELECT, intern
-#define OPTION_WARNINGS         (LL(1) << 13)      // THD, user
-#define OPTION_AUTO_IS_NULL     (LL(1) << 14)      // THD, user, binlog
-#define OPTION_FOUND_COMMENT    (LL(1) << 15)      // SELECT, intern, parser
-#define OPTION_SAFE_UPDATES     (LL(1) << 16)      // THD, user
-#define OPTION_BUFFER_RESULT    (LL(1) << 17)      // SELECT, user
-#define OPTION_BIN_LOG          (LL(1) << 18)      // THD, user
-#define OPTION_NOT_AUTOCOMMIT   (LL(1) << 19)      // THD, user
-#define OPTION_BEGIN            (LL(1) << 20)      // THD, intern
-#define OPTION_TABLE_LOCK       (LL(1) << 21)      // THD, intern
-#define OPTION_QUICK            (LL(1) << 22)      // SELECT (for DELETE)
-#define OPTION_KEEP_LOG         (LL(1) << 23)      // Keep binlog on rollback
+#define SELECT_DISTINCT          (ULL(1) << 0)       // SELECT, user
+#define SELECT_STRAIGHT_JOIN     (ULL(1) << 1)       // SELECT, user
+#define SELECT_DESCRIBE          (ULL(1) << 2)       // SELECT, user
+#define SELECT_SMALL_RESULT      (ULL(1) << 3)       // SELECT, user
+#define SELECT_BIG_RESULT        (ULL(1) << 4)       // SELECT, user
+#define OPTION_FOUND_ROWS        (ULL(1) << 5)       // SELECT, user
+#define OPTION_TO_QUERY_CACHE    (ULL(1) << 6)       // SELECT, user
+#define SELECT_NO_JOIN_CACHE     (ULL(1) << 7)       // intern
+#define OPTION_BIG_TABLES        (ULL(1) << 8)       // THD, user
+#define OPTION_BIG_SELECTS       (ULL(1) << 9)       // THD, user
+#define OPTION_LOG_OFF           (ULL(1) << 10)      // THD, user
+#define OPTION_QUOTE_SHOW_CREATE (ULL(1) << 11)      // THD, user
+#define TMP_TABLE_ALL_COLUMNS    (ULL(1) << 12)      // SELECT, intern
+#define OPTION_WARNINGS          (ULL(1) << 13)      // THD, user
+#define OPTION_AUTO_IS_NULL      (ULL(1) << 14)      // THD, user, binlog
+#define OPTION_FOUND_COMMENT     (ULL(1) << 15)      // SELECT, intern, parser
+#define OPTION_SAFE_UPDATES      (ULL(1) << 16)      // THD, user
+#define OPTION_BUFFER_RESULT     (ULL(1) << 17)      // SELECT, user
+#define OPTION_BIN_LOG           (ULL(1) << 18)      // THD, user
+#define OPTION_NOT_AUTOCOMMIT    (ULL(1) << 19)      // THD, user
+#define OPTION_BEGIN             (ULL(1) << 20)      // THD, intern
+#define OPTION_TABLE_LOCK        (ULL(1) << 21)      // THD, intern
+#define OPTION_QUICK             (ULL(1) << 22)      // SELECT (for DELETE)
+#define OPTION_KEEP_LOG          (ULL(1) << 23)      // Keep binlog on rollback
 
 /* The following is used to detect a conflict with DISTINCT */
-#define SELECT_ALL              (LL(1) << 24)      // SELECT, user, parser
+#define SELECT_ALL               (ULL(1) << 24)      // SELECT, user, parser
 
 /* Set if we are updating a non-transaction safe table */
-#define OPTION_STATUS_NO_TRANS_UPDATE   (LL(1) << 25) // THD, intern
+#define OPTION_STATUS_NO_TRANS_UPDATE   (ULL(1) << 25) // THD, intern
 
 /* The following can be set when importing tables in a 'wrong order'
    to suppress foreign key checks */
-#define OPTION_NO_FOREIGN_KEY_CHECKS    (LL(1) << 26) // THD, user, binlog
+#define OPTION_NO_FOREIGN_KEY_CHECKS    (ULL(1) << 26) // THD, user, binlog
 /* The following speeds up inserts to InnoDB tables by suppressing unique
    key checks in some cases */
-#define OPTION_RELAXED_UNIQUE_CHECKS    (LL(1) << 27) // THD, user, binlog
-#define SELECT_NO_UNLOCK                (LL(1) << 28) // SELECT, intern
-#define OPTION_SCHEMA_TABLE             (LL(1) << 29) // SELECT, intern
+#define OPTION_RELAXED_UNIQUE_CHECKS    (ULL(1) << 27) // THD, user, binlog
+#define SELECT_NO_UNLOCK                (ULL(1) << 28) // SELECT, intern
+#define OPTION_SCHEMA_TABLE             (ULL(1) << 29) // SELECT, intern
 /* Flag set if setup_tables already done */
-#define OPTION_SETUP_TABLES_DONE        (LL(1) << 30) // intern
+#define OPTION_SETUP_TABLES_DONE        (ULL(1) << 30) // intern
 /* If not set then the thread will ignore all warnings with level notes. */
-#define OPTION_SQL_NOTES                (LL(1) << 31) // THD, user
+#define OPTION_SQL_NOTES                (ULL(1) << 31) // THD, user
 /*
   Force the used temporary table to be a MyISAM table (because we will use
   fulltext functions when reading from it.
 */
-#define TMP_TABLE_FORCE_MYISAM          (LL(1) << 32)
+#define TMP_TABLE_FORCE_MYISAM          (ULL(1) << 32)
 
 /*
   Maximum length of time zone name that we support
diff --git a/sql/slave.cc b/sql/slave.cc
index 1852ff68070..493d867a618 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -31,6 +31,8 @@
 
 #include "rpl_tblmap.h"
 
+#define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
+
 #define MAX_SLAVE_RETRY_PAUSE 5
 bool use_slave_mask = 0;
 MY_BITMAP slave_error_mask;
@@ -3153,6 +3155,10 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli)
     if (!ev->when)
       ev->when = time(NULL);
     ev->thd = thd; // because up to this point, ev->thd == 0
+    DBUG_PRINT("info", ("thd->options={ %s%s}",
+                        FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
+                        FLAGSTR(thd->options, OPTION_BEGIN)));
+
     exec_res = ev->exec_event(rli);
     DBUG_PRINT("info", ("exec_event result = %d", exec_res));
     DBUG_ASSERT(rli->sql_thd==thd);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 95283ec2fc8..28d60c03c06 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -931,6 +931,8 @@ public:
     Public interface to write RBR events to the binlog
   */
   void binlog_start_trans_and_stmt();
+  int binlog_flush_transaction_cache();
+  void binlog_set_stmt_begin();
   int binlog_write_table_map(TABLE *table, bool is_transactional);
   int binlog_write_row(TABLE* table, bool is_transactional,
                        MY_BITMAP const* cols, my_size_t colcnt,
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 4dcbf9af4a0..7779adbf303 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2640,8 +2640,7 @@ void select_insert::send_error(uint errcode,const char *err)
     If the creation of the table failed (due to a syntax error, for
     example), no table will have been opened and therefore 'table'
     will be NULL. In that case, we still need to execute the rollback
-    and the end of the function to truncate the binary log, but we can
-    skip all the intermediate steps.
+    and the end of the function.
    */
   if (table)
   {
@@ -2672,10 +2671,8 @@ void select_insert::send_error(uint errcode,const char *err)
       if (!table->file->has_transactions())
       {
         if (mysql_bin_log.is_open())
-        {
           thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query, thd->query_length,
                             table->file->has_transactions(), FALSE);
-        }
         if (!thd->current_stmt_binlog_row_based && !table->s->tmp_table &&
             !can_rollback_data())
           thd->options|= OPTION_STATUS_NO_TRANS_UPDATE;
@@ -2943,6 +2940,24 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 
   TABLEOP_HOOKS *hook_ptr= NULL;
 #ifdef HAVE_ROW_BASED_REPLICATION
+  /*
+    For row-based replication, the CREATE-SELECT statement is written
+    in two pieces: the first one contain the CREATE TABLE statement
+    necessary to create the table and the second part contain the rows
+    that should go into the table.
+
+    For non-temporary tables, the start of the CREATE-SELECT
+    implicitly commits the previous transaction, and all events
+    forming the statement will be stored the transaction cache. At end
+    of the statement, the entire statement is committed as a
+    transaction, and all events are written to the binary log.
+
+    On the master, the table is locked for the duration of the
+    statement, but since the CREATE part is replicated as a simple
+    statement, there is no way to lock the table for accesses on the
+    slave.  Hence, we have to hold on to the CREATE part of the
+    statement until the statement has finished.
+   */
   class MY_HOOKS : public TABLEOP_HOOKS {
   public:
     MY_HOOKS(select_create *x) : ptr(x) { }
@@ -2952,7 +2967,7 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     {
       TABLE const *const table = *tables;
       if (ptr->get_thd()->current_stmt_binlog_row_based  &&
-          table->s->tmp_table == NO_TMP_TABLE &&
+          !table->s->tmp_table &&
           !ptr->get_create_info()->table_existed)
       {
         ptr->binlog_show_create_table(tables, count);
@@ -2970,9 +2985,9 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 
 #ifdef HAVE_ROW_BASED_REPLICATION
   /*
-    Start a statement transaction before the create if we are creating
-    a non-temporary table and are using row-based replication for the
-    statement.
+    Start a statement transaction before the create if we are using
+    row-based replication for the statement.  If we are creating a
+    temporary table, we need to start a statement transaction.
   */
   if ((thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) == 0 &&
       thd->current_stmt_binlog_row_based)
@@ -3076,13 +3091,35 @@ void select_create::store_values(List<Item> &values)
 
 void select_create::send_error(uint errcode,const char *err)
 {
+  DBUG_ENTER("select_create::send_error");
+
+  DBUG_PRINT("info",
+             ("Current statement %s row-based",
+              thd->current_stmt_binlog_row_based ? "is" : "is NOT"));
+  DBUG_PRINT("info",
+             ("Current table (at 0x%lu) %s a temporary (or non-existant) table",
+              table,
+              table && !table->s->tmp_table ? "is NOT" : "is"));
+  DBUG_PRINT("info",
+             ("Table %s prior to executing this statement",
+              get_create_info()->table_existed ? "existed" : "did not exist"));
+
   /*
-   Disable binlog, because we "roll back" partial inserts in ::abort
-   by removing the table, even for non-transactional tables.
+    This will execute any rollbacks that are necessary before writing
+    the transcation cache.
+
+    We disable the binary log since nothing should be written to the
+    binary log.  This disabling is important, since we potentially do
+    a "roll back" of non-transactional tables by removing the table,
+    and the actual rollback might generate events that should not be
+    written to the binary log.
+
   */
   tmp_disable_binlog(thd);
   select_insert::send_error(errcode, err);
   reenable_binlog(thd);
+
+  DBUG_VOID_RETURN;
 }
 
 
@@ -3093,6 +3130,14 @@ bool select_create::send_eof()
     abort();
   else
   {
+    /*
+      Do an implicit commit at end of statement for non-temporary
+      tables.  This can fail, but we should unlock the table
+      nevertheless.
+    */
+    if (!table->s->tmp_table)
+      ha_commit(thd);               // Can fail, but we proceed anyway
+
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
     VOID(pthread_mutex_lock(&LOCK_open));
@@ -3111,12 +3156,31 @@ bool select_create::send_eof()
 
 void select_create::abort()
 {
+  DBUG_ENTER("select_create::abort");
   VOID(pthread_mutex_lock(&LOCK_open));
+
+  /*
+    We roll back the statement, including truncating the transaction
+    cache of the binary log, if the statement failed.
+
+    We roll back the statement prior to deleting the table and prior
+    to releasing the lock on the table, since there might be potential
+    for failure if the rollback is executed after the drop or after
+    unlocking the table.
+
+    We also roll back the statement regardless of whether the creation
+    of the table succeeded or not, since we need to reset the binary
+    log state.
+  */
+  if (thd->current_stmt_binlog_row_based)
+    ha_rollback_stmt(thd);
+
   if (thd->extra_lock)
   {
     mysql_unlock_tables(thd, thd->extra_lock);
     thd->extra_lock=0;
   }
+
   if (table)
   {
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
@@ -3128,17 +3192,8 @@ void select_create::abort()
       table->s->version= 0;
       hash_delete(&open_cache,(byte*) table);
       if (!create_info->table_existed)
-      {
         quick_rm_table(table_type, create_table->db,
                        create_table->table_name, 0);
-        /*
-          We roll back the statement, including truncating the
-          transaction cache of the binary log, if the statement
-          failed.
-         */
-        if (thd->current_stmt_binlog_row_based)
-          ha_rollback_stmt(thd);
-      }
       /* Tell threads waiting for refresh that something has happened */
       if (version != refresh_version)
         broadcast_refresh();
@@ -3148,6 +3203,7 @@ void select_create::abort()
     table=0;                                    // Safety
   }
   VOID(pthread_mutex_unlock(&LOCK_open));
+  DBUG_VOID_RETURN;
 }