Fix for MDEV-5589: "Discrepancy in binlog on half-failed CREATE OR REPLACE"

Now if CREATE OR REPLACE fails but we have deleted a table already, we will generate a DROP TABLE in the binary log.
This fixes this issue.

In addition, for a failing CREATE OR REPLACE TABLE ... SELECT we don't generate a log of all the inserted rows, only the DROP TABLE.

I added code for not logging DROP TEMPORARY TABLE for tables where the CREATE TABLE was not logged. This code will be activated in 10.1
by removing the code protected by DONT_LOG_DROP_OF_TEMPORARY_TABLES.





mysql-test/suite/rpl/r/create_or_replace_mix.result:
  More test cases
mysql-test/suite/rpl/r/create_or_replace_row.result:
  More test cases
mysql-test/suite/rpl/r/create_or_replace_statement.result:
  More test cases
mysql-test/suite/rpl/t/create_or_replace.inc:
  More test cases
sql/log.cc:
  Added binlog_reset_cache() to clear the binary log.
sql/log.h:
  Added prototype
sql/sql_insert.cc:
  If CREATE OR REPLACE TABLE ... SELECT fails:
  - Don't log anything if nothing changed
  - If table was deleted, log a DROP TABLE.
  Remember if we table creation of temporary tables was logged.
sql/sql_table.cc:
  Added log_drop_table()
  Remember if we table creation of temporary tables was logged.
  If CREATE OR REPLACE TABLE ... SELECT fails and a table was deleted, log a DROP TABLE.
sql/sql_table.h:
  Added prototype
sql/sql_truncate.cc:
  Remember if we table creation of temporary tables was logged.
sql/table.h:
  Added table_creation_was_logged

11 files changed, 335 insertions, 30 deletions

diff --git a/mysql-test/suite/rpl/r/create_or_replace_mix.result b/mysql-test/suite/rpl/r/create_or_replace_mix.result
index 99de4ba729d..88837ebbf46 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_mix.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_mix.result
@@ -67,6 +67,9 @@ ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+create temporary table t9 (a int);
+create or replace temporary table t9 (a int primary key) select a from t2;
+ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 binlog from server 1
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -79,20 +82,27 @@ master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
-  `a` int(11) NOT NULL,
-  PRIMARY KEY (`a`)
-)
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
+master-bin.000001	#	Query	#	#	ROLLBACK
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t9`/* Generated to handle failed CREATE OR REPLACE */
 master-bin.000001	#	Query	#	#	ROLLBACK
 show tables;
 Tables_in_test
-t1
 t2
+create table t1 (a int);
+create or replace table t1 (a int, a int) select * from t2;
+ERROR 42S21: Duplicate column name 'a'
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
 Warnings:
 Note	1051	Unknown table 'test.t1'
+drop temporary table if exists t9;
+Warnings:
+Note	1051	Unknown table 'test.t9'
 #
 # Ensure that CREATE are run as CREATE OR REPLACE on slave
 #
@@ -158,5 +168,39 @@ slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
 slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 slave-bin.000001	#	Xid	#	#	COMMIT /* XID */
 drop table t1;
-drop table t2,t3;
+#
+# Check logging of drop temporary table
+#
+drop temporary table t3;
+set @org_binlog_format=@@binlog_format;
+set binlog_format="STATEMENT";
+create temporary table t5 (a int);
+drop temporary table t5;
+set binlog_format="ROW";
+create temporary table t6 (a int);
+drop temporary table t6;
+set binlog_format="STATEMENT";
+create temporary table t7 (a int);
+set binlog_format="ROW";
+drop temporary table t7;
+create temporary table t8 (a int);
+set binlog_format="STATEMENT";
+ERROR HY000: Cannot switch out of the row-based binary log format when the session has open temporary tables
+drop temporary table t8;
+set @@binlog_format=@org_binlog_format;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t5 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TEMPORARY TABLE `t5` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t6` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t7 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t7` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t8` /* generated by server */
+drop table t2;
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/create_or_replace_row.result b/mysql-test/suite/rpl/r/create_or_replace_row.result
index 7c3a27573e5..9921ece7588 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_row.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_row.result
@@ -89,6 +89,9 @@ ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+create temporary table t9 (a int);
+create or replace temporary table t9 (a int primary key) select a from t2;
+ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 binlog from server 1
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -101,20 +104,27 @@ master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
-  `a` int(11) NOT NULL,
-  PRIMARY KEY (`a`)
-)
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
+master-bin.000001	#	Query	#	#	ROLLBACK
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t9`/* Generated to handle failed CREATE OR REPLACE */
 master-bin.000001	#	Query	#	#	ROLLBACK
 show tables;
 Tables_in_test
-t1
 t2
+create table t1 (a int);
+create or replace table t1 (a int, a int) select * from t2;
+ERROR 42S21: Duplicate column name 'a'
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
 Warnings:
 Note	1051	Unknown table 'test.t1'
+drop temporary table if exists t9;
+Warnings:
+Note	1051	Unknown table 'test.t9'
 #
 # Ensure that CREATE are run as CREATE OR REPLACE on slave
 #
@@ -180,5 +190,39 @@ slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
 slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 slave-bin.000001	#	Xid	#	#	COMMIT /* XID */
 drop table t1;
-drop table t2,t3;
+#
+# Check logging of drop temporary table
+#
+drop temporary table t3;
+set @org_binlog_format=@@binlog_format;
+set binlog_format="STATEMENT";
+create temporary table t5 (a int);
+drop temporary table t5;
+set binlog_format="ROW";
+create temporary table t6 (a int);
+drop temporary table t6;
+set binlog_format="STATEMENT";
+create temporary table t7 (a int);
+set binlog_format="ROW";
+drop temporary table t7;
+create temporary table t8 (a int);
+set binlog_format="STATEMENT";
+ERROR HY000: Cannot switch out of the row-based binary log format when the session has open temporary tables
+drop temporary table t8;
+set @@binlog_format=@org_binlog_format;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t5 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TEMPORARY TABLE `t5` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t6` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t7 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t7` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t8` /* generated by server */
+drop table t2;
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/create_or_replace_statement.result b/mysql-test/suite/rpl/r/create_or_replace_statement.result
index f5d77a0f697..bb799848185 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_statement.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_statement.result
@@ -67,6 +67,9 @@ ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+create temporary table t9 (a int);
+create or replace temporary table t9 (a int primary key) select a from t2;
+ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 binlog from server 1
 include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -79,13 +82,27 @@ master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; create or replace table t1 (a int primary key) select a from t2
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t9 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t9`/* Generated to handle failed CREATE OR REPLACE */
 show tables;
 Tables_in_test
 t2
+create table t1 (a int);
+create or replace table t1 (a int, a int) select * from t2;
+ERROR 42S21: Duplicate column name 'a'
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
 Warnings:
 Note	1051	Unknown table 'test.t1'
+drop temporary table if exists t9;
+Warnings:
+Note	1051	Unknown table 'test.t9'
 #
 # Ensure that CREATE are run as CREATE OR REPLACE on slave
 #
@@ -140,5 +157,39 @@ slave-bin.000001	#	Query	#	#	use `test`; create table t2 engine=myisam select *
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
 slave-bin.000001	#	Query	#	#	use `test`; create or replace table t2 engine=innodb select * from t1
 drop table t1;
-drop table t2,t3;
+#
+# Check logging of drop temporary table
+#
+drop temporary table t3;
+set @org_binlog_format=@@binlog_format;
+set binlog_format="STATEMENT";
+create temporary table t5 (a int);
+drop temporary table t5;
+set binlog_format="ROW";
+create temporary table t6 (a int);
+drop temporary table t6;
+set binlog_format="STATEMENT";
+create temporary table t7 (a int);
+set binlog_format="ROW";
+drop temporary table t7;
+create temporary table t8 (a int);
+set binlog_format="STATEMENT";
+ERROR HY000: Cannot switch out of the row-based binary log format when the session has open temporary tables
+drop temporary table t8;
+set @@binlog_format=@org_binlog_format;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t5 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TEMPORARY TABLE `t5` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t6` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t7 (a int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t7` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t8` /* generated by server */
+drop table t2;
 include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/create_or_replace.inc b/mysql-test/suite/rpl/t/create_or_replace.inc
index dad705403ed..b7ba4bc2ba6 100644
--- a/mysql-test/suite/rpl/t/create_or_replace.inc
+++ b/mysql-test/suite/rpl/t/create_or_replace.inc
@@ -50,10 +50,15 @@ drop table if exists t1;
 create or replace table t1 (a int primary key) select a from t2;
 
 create table t1 (a int);
-# This should be logged as we will delete t1
+# This should as a delete as we will delete t1
 --error ER_DUP_ENTRY
 create or replace table t1 (a int primary key) select a from t2;
 
+# Same with temporary table
+create temporary table t9 (a int);
+--error ER_DUP_ENTRY
+create or replace temporary table t9 (a int primary key) select a from t2;
+
 --echo binlog from server 1
 --source include/show_binlog_events.inc
 save_master_pos;
@@ -62,7 +67,14 @@ sync_with_master;
 show tables;
 connection server_1;
 
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+create table t1 (a int);
+--error ER_DUP_FIELDNAME
+create or replace table t1 (a int, a int) select * from t2;
+--source include/show_binlog_events.inc
+
 drop table if exists t1,t2;
+drop temporary table if exists t9;
 
 --echo #
 --echo # Ensure that CREATE are run as CREATE OR REPLACE on slave
@@ -131,7 +143,34 @@ sync_with_master;
 connection server_1;
 drop table t1;
 
+--echo #
+--echo # Check logging of drop temporary table
+--echo #
+
+drop temporary table t3;
+
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+
+set @org_binlog_format=@@binlog_format;
+set binlog_format="STATEMENT";
+create temporary table t5 (a int);
+drop temporary table t5;
+set binlog_format="ROW";
+create temporary table t6 (a int);
+drop temporary table t6;
+set binlog_format="STATEMENT";
+create temporary table t7 (a int);
+set binlog_format="ROW";
+drop temporary table t7;
+create temporary table t8 (a int);
+--error ER_TEMP_TABLE_PREVENTS_SWITCH_OUT_OF_RBR
+set binlog_format="STATEMENT";
+drop temporary table t8;
+set @@binlog_format=@org_binlog_format;
+
+--source include/show_binlog_events.inc
+
 # Clean up
-drop table t2,t3;
+drop table t2;
 
 --source include/rpl_end.inc
diff --git a/sql/log.cc b/sql/log.cc
index ebfbba953fa..1943be2817f 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -2052,6 +2052,19 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   DBUG_RETURN(error);
 }
 
+
+void binlog_reset_cache(THD *thd)
+{
+  binlog_cache_mngr *const cache_mngr= 
+    (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
+  DBUG_ENTER("binlog_reset_cache");
+  thd->binlog_remove_pending_rows_event(TRUE, TRUE);
+  cache_mngr->reset(true, true);
+  thd->clear_binlog_table_maps();
+  DBUG_VOID_RETURN;
+}
+
+
 void MYSQL_BIN_LOG::set_write_error(THD *thd, bool is_transactional)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::set_write_error");
diff --git a/sql/log.h b/sql/log.h
index 2577cc5225a..67fcf068ec4 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -1012,6 +1012,7 @@ File open_binlog(IO_CACHE *log, const char *log_file_name,
                  const char **errmsg);
 
 void make_default_log_name(char **out, const char* log_ext, bool once);
+void binlog_reset_cache(THD *thd);
 
 extern MYSQL_PLUGIN_IMPORT MYSQL_BIN_LOG mysql_bin_log;
 extern LOGGER logger;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 92be2296f6f..e258ac14776 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3960,6 +3960,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
         */
         DBUG_ASSERT(0);
       }
+      DBUG_ASSERT(create_table->table == create_info->table);
     }
   }
   else
@@ -4247,6 +4248,8 @@ bool select_create::send_eof()
     if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
       trans_commit_implicit(thd);
   }
+  else if (!thd->is_current_stmt_binlog_format_row())
+    table->s->table_creation_was_logged= 1;
 
   table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
@@ -4310,8 +4313,7 @@ void select_create::abort_result_set()
   */
 
   save_option_bits= thd->variables.option_bits;
-  if (!(thd->log_current_statement))
-    thd->variables.option_bits&= ~OPTION_BIN_LOG;
+  thd->variables.option_bits&= ~OPTION_BIN_LOG;
   select_insert::abort_result_set();
   thd->transaction.stmt.modified_non_trans_table= FALSE;
   thd->variables.option_bits= save_option_bits;
@@ -4334,11 +4336,21 @@ void select_create::abort_result_set()
 
   if (table)
   {
+    bool tmp_table= table->s->tmp_table;
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
     table->auto_increment_field_not_null= FALSE;
     drop_open_table(thd, table, create_table->db, create_table->table_name);
     table=0;                                    // Safety
+    if (thd->log_current_statement)
+    {
+      /* Remove logging of drop, create + insert rows */
+      binlog_reset_cache(thd);
+      /* Original table was deleted. We have to log it */
+      log_drop_table(thd, create_table->db, create_table->db_length,
+                     create_table->table_name, create_table->table_name_length,
+                     tmp_table);
+    }
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8e9601c437e..a815ec2ea8f 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2288,6 +2288,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   for (table= tables; table; table= table->next_local)
   {
     bool is_trans= 0;
+    bool table_creation_was_logged= 1;
     char *db=table->db;
     size_t db_length= table->db_length;
     handlerton *table_type= 0;
@@ -2316,6 +2317,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       error= 1;
     else
     {
+      table_creation_was_logged= table->table->s->table_creation_was_logged;
       if ((error= drop_temporary_table(thd, table->table, &is_trans)) == -1)
       {
         DBUG_ASSERT(thd->in_sub_stmt);
@@ -2336,7 +2338,10 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           . "DROP" was executed but a temporary table was affected (.i.e
           !error).
       */
-      if (!dont_log_query)
+#ifndef DONT_LOG_DROP_OF_TEMPORARY_TABLES
+      table_creation_was_logged= 1;
+#endif
+      if (!dont_log_query && table_creation_was_logged)
       {
         /*
           If there is an error, we don't know the type of the engine
@@ -2660,6 +2665,43 @@ end:
   DBUG_RETURN(error);
 }
 
+/**
+  Log the drop of a table.
+
+  @param thd	           Thread handler
+  @param db_name           Database name
+  @param table_name        Table name
+  @param temporary_table   1 if table was a temporary table
+
+  This code is only used in the case of failed CREATE OR REPLACE TABLE
+  when the original table was dropped but we could not create the new one.
+*/
+
+bool log_drop_table(THD *thd, const char *db_name, size_t db_name_length,
+                    const char *table_name, size_t table_name_length,
+                    bool temporary_table)
+{
+  char buff[NAME_LEN*2 + 80];
+  String query(buff, sizeof(buff), system_charset_info);
+  bool error;
+  DBUG_ENTER("log_drop_table");
+
+  query.length(0);
+  query.append(STRING_WITH_LEN("DROP "));
+  if (temporary_table)
+    query.append(STRING_WITH_LEN("TEMPORARY "));
+  query.append(STRING_WITH_LEN("TABLE IF EXISTS "));
+  append_identifier(thd, &query, db_name, db_name_length);
+  query.append(".");
+  append_identifier(thd, &query, table_name, table_name_length);
+  query.append(STRING_WITH_LEN("/* Generated to handle "
+                               "failed CREATE OR REPLACE */"));
+  error= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                           query.ptr(), query.length(),
+                           FALSE, FALSE, temporary_table, 0);
+  DBUG_RETURN(error);
+}
+
 
 /**
   Quickly remove a table.
@@ -4590,6 +4632,7 @@ int create_table_impl(THD *thd,
     TABLE *tmp_table;
     if ((tmp_table= find_temporary_table(thd, db, table_name)))
     {
+      bool table_creation_was_logged= tmp_table->s->table_creation_was_logged;
       if (create_info->options & HA_LEX_CREATE_REPLACE)
       {
         bool is_trans;
@@ -4607,6 +4650,19 @@ int create_table_impl(THD *thd,
         my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
         goto err;
       }
+      /*
+        We have to log this query, even if it failed later to ensure the
+        drop is done.
+      */
+#ifndef DONT_LOG_DROP_OF_TEMPORARY_TABLES
+      table_creation_was_logged= 1;
+#endif
+      if (table_creation_was_logged)
+      {
+        thd->variables.option_bits|= OPTION_KEEP_LOG;
+        thd->log_current_statement= 1;
+        create_info->table_was_deleted= 1;
+      }
     }
   }
   else
@@ -4720,6 +4776,7 @@ int create_table_impl(THD *thd,
       goto err;
   }
 
+  create_info->table= 0;
   if (!frm_only && create_info->tmp_table())
   {
     /*
@@ -4740,6 +4797,7 @@ int create_table_impl(THD *thd,
       *is_trans= table->file->has_transactions();
 
     thd->thread_specific_used= TRUE;
+    create_info->table= table;                  // Store pointer to table
   }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   else if (thd->work_part_info && frm_only)
@@ -4912,6 +4970,7 @@ err:
   /* In RBR we don't need to log CREATE TEMPORARY TABLE */
   if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
     DBUG_RETURN(result);
+
   /* Write log if no error or if we already deleted a table */
   if (!result || thd->log_current_statement)
   {
@@ -4922,7 +4981,15 @@ err:
         associated with it and do UNLOCK_TABLES if no more locked tables.
       */
       thd->locked_tables_list.unlock_locked_table(thd, mdl_ticket);
-    }    
+    }
+    else if (!result && create_info->tmp_table() && create_info->table)
+    {
+      /*
+        Remember that tmp table creation was logged so that we know if
+        we should log a delete of it.
+      */
+      create_info->table->s->table_creation_was_logged= 1;
+    }
     if (write_bin_log(thd, result ? FALSE : TRUE, thd->query(),
                       thd->query_length(), is_trans))
       result= 1;
@@ -5356,13 +5423,38 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
     */
   }
   else
+  {
+    DBUG_PRINT("info",
+               ("res: %d  tmp_table: %d  create_info->table: %p",
+                res, create_info->tmp_table(), local_create_info.table));
+    if (!res && create_info->tmp_table() && local_create_info.table)
+    {
+      /*
+        Remember that tmp table creation was logged so that we know if
+        we should log a delete of it.
+      */
+      local_create_info.table->s->table_creation_was_logged= 1;
+    }
     do_logging= TRUE;
+  }
 
 err:
-  if (do_logging &&
-      write_bin_log(thd, res ? FALSE : TRUE, thd->query(),
-                    thd->query_length(), is_trans))
-    res= 1;
+  if (do_logging)
+  {
+    if (res && create_info->table_was_deleted)
+    {
+      /*
+        Table was not deleted. Original table was deleted.
+        We have to log it.
+      */
+      log_drop_table(thd, table->db, table->db_length,
+                     table->table_name, table->table_name_length,
+                     create_info->tmp_table());
+    }
+    else if (write_bin_log(thd, res ? FALSE : TRUE, thd->query(),
+                           thd->query_length(), is_trans))
+      res= 1;
+  }
   DBUG_RETURN(res);
 }
 
@@ -8718,6 +8810,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
         mysql_lock_remove(thd, thd->lock, table);
       }
     }
+    new_table->s->table_creation_was_logged=
+      table->s->table_creation_was_logged;
     /* Remove link to old table and rename the new one */
     close_temporary_table(thd, table, true, true);
     /* Should pass the 'new_name' as we store table name in the cache */
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 7255ce68743..c148ae63b4e 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -240,6 +240,9 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                             bool drop_temporary, bool drop_view,
                             bool log_query, bool dont_free_locks);
+bool log_drop_table(THD *thd, const char *db_name, size_t db_name_length,
+                    const char *table_name, size_t table_name_length,
+                    bool temporary_table);
 bool quick_rm_table(THD *thd, handlerton *base, const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index 907996c2315..0ed276cbd23 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -256,6 +256,7 @@ static bool recreate_temporary_table(THD *thd, TABLE *table)
   bool error= TRUE;
   TABLE_SHARE *share= table->s;
   handlerton *table_type= table->s->db_type();
+  TABLE *new_table;
   DBUG_ENTER("recreate_temporary_table");
 
   table->file->info(HA_STATUS_AUTO | HA_STATUS_NO_LOCK);
@@ -266,11 +267,13 @@ static bool recreate_temporary_table(THD *thd, TABLE *table)
   dd_recreate_table(thd, share->db.str, share->table_name.str,
                     share->normalized_path.str);
 
-  if (open_table_uncached(thd, table_type, share->path.str, share->db.str,
-                          share->table_name.str, true, true))
+  if ((new_table= open_table_uncached(thd, table_type, share->path.str,
+                                      share->db.str,
+                                      share->table_name.str, true, true)))
   {
     error= FALSE;
     thd->thread_specific_used= TRUE;
+    new_table->s->table_creation_was_logged= share->table_creation_was_logged;
   }
   else
     rm_temporary_table(table_type, share->path.str);
diff --git a/sql/table.h b/sql/table.h
index 043760dd887..fa88a2c4cfb 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -742,6 +742,7 @@ struct TABLE_SHARE
   bool is_view;
   bool deleting;                        /* going to delete this table */
   bool can_cmp_whole_record;
+  bool table_creation_was_logged;
   ulong table_map_id;                   /* for row-based replication */
 
   /*