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

7 files changed, 137 insertions, 10 deletions

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 */
 
   /*