Fix for BUG#2477 "Slave stop with error after master reboot if use HEAP tables":

when we open the HEAP table for the first time since server restart,
in hp_open(), we set a flag to propagate this info to the handler level
which then writes a DELETE FROM this_heap_table to the binlog.
It is not a perfect solution for the bug, because between the server start and 
the first open of the table, the slave still had old data in his table so
a SELECT on the slave may show wrong content. But if there is a --init-file
to populate the HEAP table on master as startup, then this is a safe fix
(I'll put a note about init-file in the HEAP section of the manual).

8 files changed, 136 insertions, 18 deletions

diff --git a/heap/hp_info.c b/heap/hp_info.c
index 3e9d6b6a90b..c8cdfc58c2d 100644
--- a/heap/hp_info.c
+++ b/heap/hp_info.c
@@ -48,12 +48,13 @@ int heap_info(reg1 HP_INFO *info,reg2 HEAPINFO *x,
 	      int flag __attribute__((unused)))
 {
   DBUG_ENTER("heap_info");
-  x->records	 = info->s->records;
-  x->deleted	 = info->s->deleted;
-  x->reclength	 = info->s->reclength;
-  x->data_length = info->s->data_length;
-  x->index_length= info->s->index_length;
-  x->max_records = info->s->max_records;
-  x->errkey	 = info->errkey;
+  x->records         = info->s->records;
+  x->deleted         = info->s->deleted;
+  x->reclength       = info->s->reclength;
+  x->data_length     = info->s->data_length;
+  x->index_length    = info->s->index_length;
+  x->max_records     = info->s->max_records;
+  x->errkey          = info->errkey;
+  x->implicit_emptied= info->implicit_emptied;
   DBUG_RETURN(0);
 } /* heap_info */
diff --git a/heap/hp_open.c b/heap/hp_open.c
index ec8b5fea904..3bf2881667a 100644
--- a/heap/hp_open.c
+++ b/heap/hp_open.c
@@ -30,6 +30,7 @@ HP_INFO *heap_open(const char *name, int mode, uint keys, HP_KEYDEF *keydef,
 		   uint reclength, ulong max_records, ulong min_records)
 {
   uint i,j,key_segs,max_length,length;
+  my_bool implicit_emptied= 0;
   HP_INFO *info;
   HP_SHARE *share;
   HP_KEYSEG *keyseg;
@@ -39,6 +40,7 @@ HP_INFO *heap_open(const char *name, int mode, uint keys, HP_KEYDEF *keydef,
   if (!(share=_hp_find_named_heap(name)))
   {
     DBUG_PRINT("info",("Initializing new table"));
+    implicit_emptied= 1;
     for (i=key_segs=max_length=0 ; i < keys ; i++)
     {
       key_segs+= keydef[i].keysegs;
@@ -127,6 +129,7 @@ HP_INFO *heap_open(const char *name, int mode, uint keys, HP_KEYDEF *keydef,
 #ifndef DBUG_OFF
   info->opt_flag=READ_CHECK_USED;		/* Check when changing */
 #endif
+  info->implicit_emptied= implicit_emptied;
   DBUG_PRINT("exit",("heap: %lx  reclength: %d  records_in_block: %d",
 		     info,share->reclength,share->block.records_in_block));
   DBUG_RETURN(info);
diff --git a/include/heap.h b/include/heap.h
index 02b04e2b3ec..ed023133469 100644
--- a/include/heap.h
+++ b/include/heap.h
@@ -47,6 +47,7 @@ typedef struct st_heapinfo		/* Struct from heap_info */
   ulong index_length;
   uint reclength;			/* Length of one record */
   int errkey;
+  my_bool implicit_emptied;
 } HEAPINFO;
 
 
@@ -126,6 +127,7 @@ typedef struct st_heap_info
   int  mode;				/* Mode of file (READONLY..) */
   uint opt_flag,update;
   byte *lastkey;			/* Last used key with rkey */
+  my_bool implicit_emptied;
 #ifdef THREAD
   THR_LOCK_DATA lock;
 #endif
diff --git a/mysql-test/r/rpl_heap.result b/mysql-test/r/rpl_heap.result
new file mode 100644
index 00000000000..1556bcd5f25
--- /dev/null
+++ b/mysql-test/r/rpl_heap.result
@@ -0,0 +1,29 @@
+reset master;
+drop table if exists t1;
+create table t1 (a int) type=HEAP;
+insert into t1 values(10);
+show binlog events from 79;
+Log_name	Pos	Event_type	Server_id	Orig_log_pos	Info
+master-bin.001	79	Query	1	79	use `test`; create table t1 (a int) type=HEAP
+master-bin.001	147	Query	1	147	use `test`; DELETE FROM `test`.`t1`
+master-bin.001	205	Query	1	205	use `test`; insert into t1 values(10)
+reset slave;
+start slave;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) default NULL
+) TYPE=HEAP
+select * from t1;
+a
+10
+select * from t1;
+a
+select * from t1 limit 10;
+a
+show binlog events in 'master-bin.002' from 79;
+Log_name	Pos	Event_type	Server_id	Orig_log_pos	Info
+master-bin.002	79	Query	1	79	use `test`; DELETE FROM `test`.`t1`
+select * from t1;
+a
+drop table t1;
diff --git a/mysql-test/t/rpl_heap.test b/mysql-test/t/rpl_heap.test
new file mode 100644
index 00000000000..15f61918034
--- /dev/null
+++ b/mysql-test/t/rpl_heap.test
@@ -0,0 +1,47 @@
+# You must run this test with --manager.
+
+require_manager;
+
+# Don't know why, but using TCP/IP connections makes this test fail
+# with "Lost connection to MySQL server during query" when we
+# issue a query after the server restart.
+# Maybe this is something awkward in mysqltest or in the manager?
+# So we use sockets.
+connect (master,localhost,root,,test,0,master.sock);
+connect (slave,localhost,root,,test,0,slave.sock);
+
+connection master;
+reset master;
+drop table if exists t1;
+create table t1 (a int) type=HEAP;
+insert into t1 values(10);
+save_master_pos;
+show binlog events from 79;
+connection slave;
+reset slave;
+start slave;
+sync_with_master;
+show create table t1;
+select * from t1; # should be one row
+
+server_stop master;
+server_start master;
+
+connection master;
+select * from t1;
+# to check that DELETE is not written twice
+# (the LIMIT is to not use the query cache)
+select * from t1 limit 10;
+save_master_pos;
+show binlog events in 'master-bin.002' from 79;
+
+connection slave;
+sync_with_master;
+select * from t1; # should be empty
+
+# clean up
+connection master;
+drop table t1;
+save_master_pos;
+connection slave;
+sync_with_master;
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index 2edc3b1478e..eb4bf517374 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -222,6 +222,7 @@ void ha_heap::info(uint flag)
   index_file_length=info.index_length;
   max_data_file_length= info.max_records* info.reclength;
   delete_length= info.deleted * info.reclength;
+  implicit_emptied= info.implicit_emptied;
 }
 
 int ha_heap::extra(enum ha_extra_function operation)
diff --git a/sql/handler.h b/sql/handler.h
index 8e72267c337..60edf539e2c 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -213,6 +213,7 @@ public:
   uint raid_type,raid_chunks;
   FT_INFO *ft_handler;
   bool  auto_increment_column_changed;
+  bool implicit_emptied;                /* Can be !=0 only if HEAP */
 
   handler(TABLE *table_arg) :table(table_arg),
     ref(0), data_file_length(0), max_data_file_length(0), index_file_length(0),
@@ -221,7 +222,7 @@ public:
     create_time(0), check_time(0), update_time(0),
     key_used_on_scan(MAX_KEY), active_index(MAX_REF_PARTS),
     ref_length(sizeof(my_off_t)), block_size(0),
-    raid_type(0), ft_handler(0)
+    raid_type(0), ft_handler(0), implicit_emptied(0)
     {}
   virtual ~handler(void) {}
   int ha_open(const char *name, int mode, int test_if_locked);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 9937336fb4a..40153ad847e 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -550,6 +550,11 @@ void close_temporary_tables(THD *thd)
   query_buf_size= 50;   // Enough for DROP ... TABLE
 
   for (table=thd->temporary_tables ; table ; table=table->next)
+    /*
+      We are going to add 4 ` around the db/table names, so 1 does not look
+      enough; indeed it is enough, because table->key_length is greater (by 8,
+      because of server_id and thread_id) than db||table.
+    */
     query_buf_size+= table->key_length+1;
 
   if ((query = alloc_root(&thd->mem_root, query_buf_size)))
@@ -566,8 +571,8 @@ void close_temporary_tables(THD *thd)
         Here we assume table_cache_key always starts
         with \0 terminated db name
       */
-      end = strxmov(end,"`",table->table_cache_key,"`",
-                    ".`",table->real_name,"`,", NullS);
+      end = strxmov(end,"`",table->table_cache_key,"`.`",
+                    table->real_name,"`,", NullS);
     }
     next=table->next;
     close_temporary(table);
@@ -1331,8 +1336,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       }
     }
     pthread_mutex_unlock(&LOCK_open);
-    thd->net.last_error[0]=0;				// Clear error message
-    thd->net.last_errno=0;
+    thd->clear_error();
     error=0;
     if (openfrm(path,alias,
 		(uint) (HA_OPEN_KEYFILE | HA_OPEN_RNDFILE | HA_GET_INDEX |
@@ -1343,8 +1347,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
 	(entry->file->is_crashed() && entry->file->check_and_repair(thd)))
     {
       /* Give right error message */
-      thd->net.last_error[0]=0;
-      thd->net.last_errno=0;
+      thd->clear_error();
       my_error(ER_NOT_KEYFILE, MYF(0), name, my_errno);
       sql_print_error("Error: Couldn't repair table: %s.%s",db,name);
       if (entry->file)
@@ -1352,16 +1355,47 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       error=1;
     }
     else
-    {
-      thd->net.last_error[0]=0;			// Clear error message
-      thd->net.last_errno=0;
-    }
+      thd->clear_error();
     pthread_mutex_lock(&LOCK_open);
     unlock_table_name(thd,&table_list);
 
     if (error)
       goto err;
   }
+  /*
+    If we are here, there was no fatal error (but error may be still
+    unitialized).
+  */
+  if (unlikely(entry->file->implicit_emptied))
+  {
+    entry->file->implicit_emptied= 0;
+    if (mysql_bin_log.is_open())
+    {
+      char *query, *end;
+      uint query_buf_size= 20 + 2*NAME_LEN + 1;
+      if ((query= (char*)my_malloc(query_buf_size,MYF(MY_WME))))
+      {
+        end = strxmov(strmov(query, "DELETE FROM `"),
+                      db,"`.`",name,"`", NullS);
+        Query_log_event qinfo(thd, query, (ulong)(end-query), 0);
+        mysql_bin_log.write(&qinfo);
+        my_free(query, MYF(0));
+      }
+      else
+      {
+        /*
+          As replication is maybe going to be corrupted, we need to warn the
+          DBA on top of warning the client (which will automatically be done
+          because of MYF(MY_WME) in my_malloc() above).
+        */
+        sql_print_error("Error: when opening HEAP table, could not allocate \
+memory to write 'DELETE FROM `%s`.`%s`' to the binary log",db,name);
+        if (entry->file)
+          closefrm(entry);
+        goto err;
+      }
+    }
+  }
   DBUG_RETURN(0);
 err:
   DBUG_RETURN(1);