Merge romeo.(none):/home/bk/b22864-mysql-5.1-new-rpl

into  romeo.(none):/home/bk/merge-b22864-myql-5.1-new-rpl


sql/log.h:
  Auto merged
sql/log_event.h:
  Auto merged
sql/mysql_priv.h:
  Auto merged
sql/sql_class.h:
  Auto merged
sql/log.cc:
  Merge with mysql-5.1-new-rpl
sql/slave.cc:
  Merge with mysql-5.1-new-rpl
sql/sql_insert.cc:
  Merge with mysql-5.1-new-rpl

10 files changed, 563 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 41e413e9ff7..06b8396fbd9 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 {
@@ -113,9 +148,15 @@ public:
    */
   void truncate(my_off_t pos)
   {
+#ifdef HAVE_ROW_BASED_REPLICATION
+    DBUG_PRINT("info", ("truncating to position %lu", pos));
+    DBUG_PRINT("info", ("before_stmt_pos=%lu", pos));
     delete pending();
     set_pending(0);
     reinit_io_cache(&trans_log, WRITE_CACHE, pos, 0, 0);
+    if (pos < before_stmt_pos)
+      before_stmt_pos= MY_OFF_T_UNDEF;
+#endif
   }
 
   /*
@@ -1483,12 +1524,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
@@ -2075,7 +2115,7 @@ bool MYSQL_QUERY_LOG::write(time_t event_time, const char *user_host,
         if (my_b_write(&log_file, (byte*) "\t\t" ,2) < 0)
           goto err;
 
-    /* command_type, thread_id */
+      /* command_type, thread_id */
       length= my_snprintf(buff, 32, "%5ld ", (long) thread_id);
 
     if (my_b_write(&log_file, (byte*) buff, length))
@@ -3415,18 +3455,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);
@@ -3434,6 +3463,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.
  */
@@ -3844,12 +3918,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.
@@ -3909,20 +4019,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 2316c96f266..80c97b8db39 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -340,6 +340,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 021ce938139..5a7959a01cc 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -425,12 +425,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 09cc4b5c219..d9b49d50f10 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -307,54 +307,54 @@ MY_LOCALE *my_locale_by_number(uint number);
    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 67e8ba20c4f..307503a46d8 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -33,6 +33,7 @@
 
 int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
 
+#define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
 
 #define MAX_SLAVE_RETRY_PAUSE 5
 bool use_slave_mask = 0;
@@ -1799,6 +1800,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 a5bdeee9d16..6167a86b057 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -946,6 +946,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 eca301aa1c6..8c4d2e537e5 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2646,8 +2646,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)
   {
@@ -2678,10 +2677,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;
@@ -2948,6 +2945,24 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   DBUG_ENTER("select_create::prepare");
 
   TABLEOP_HOOKS *hook_ptr= NULL;
+  /*
+    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) { }
@@ -2957,7 +2972,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);
@@ -2973,9 +2988,9 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   unit= u;
 
   /*
-    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)
@@ -3075,13 +3090,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;
 }
 
 
@@ -3092,6 +3129,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));
@@ -3110,12 +3155,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);
@@ -3127,17 +3191,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();
@@ -3147,6 +3202,7 @@ void select_create::abort()
     table=0;                                    // Safety
   }
   VOID(pthread_mutex_unlock(&LOCK_open));
+  DBUG_VOID_RETURN;
 }