Bug #38360: BLACKHOLE replication with RBR is broken

The Blackhole engine did not support row-based replication
since the delete_row(), update_row(), and the index and range
searching functions were not implemented.

This patch adds row-based replication support for the
Blackhole engine by implementing the two functions mentioned
above, and making the engine pretend that it has found the
correct row to delete or update when executed from the slave
SQL thread by implementing index and range searching functions.

It is necessary to only pretend this for the SQL thread, since
a SELECT executed on the Blackhole engine will otherwise never
return EOF, causing a livelock.

2 files changed, 36 insertions, 2 deletions

diff --git a/storage/blackhole/ha_blackhole.cc b/storage/blackhole/ha_blackhole.cc
index 85b8117e501..1a1a0d02375 100644
--- a/storage/blackhole/ha_blackhole.cc
+++ b/storage/blackhole/ha_blackhole.cc
@@ -18,6 +18,7 @@
 #pragma implementation				// gcc: Class implementation
 #endif
 
+#define MYSQL_SERVER 1
 #include "mysql_priv.h"
 #include "ha_blackhole.h"
 
@@ -100,6 +101,24 @@ int ha_blackhole::write_row(uchar * buf)
   DBUG_RETURN(table->next_number_field ? update_auto_increment() : 0);
 }
 
+int ha_blackhole::update_row(const uchar *old_data, uchar *new_data)
+{
+  DBUG_ENTER("ha_blackhole::update_row");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
+  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
+}
+
+int ha_blackhole::delete_row(const uchar *buf)
+{
+  DBUG_ENTER("ha_blackhole::delete_row");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
+  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
+}
+
 int ha_blackhole::rnd_init(bool scan)
 {
   DBUG_ENTER("ha_blackhole::rnd_init");
@@ -110,6 +129,9 @@ int ha_blackhole::rnd_init(bool scan)
 int ha_blackhole::rnd_next(uchar *buf)
 {
   DBUG_ENTER("ha_blackhole::rnd_next");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
   DBUG_RETURN(HA_ERR_END_OF_FILE);
 }
 
@@ -189,6 +211,9 @@ int ha_blackhole::index_read_map(uchar * buf, const uchar * key,
                              enum ha_rkey_function find_flag)
 {
   DBUG_ENTER("ha_blackhole::index_read");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
   DBUG_RETURN(HA_ERR_END_OF_FILE);
 }
 
@@ -198,6 +223,9 @@ int ha_blackhole::index_read_idx_map(uchar * buf, uint idx, const uchar * key,
                                  enum ha_rkey_function find_flag)
 {
   DBUG_ENTER("ha_blackhole::index_read_idx");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
   DBUG_RETURN(HA_ERR_END_OF_FILE);
 }
 
@@ -206,6 +234,9 @@ int ha_blackhole::index_read_last_map(uchar * buf, const uchar * key,
                                       key_part_map keypart_map)
 {
   DBUG_ENTER("ha_blackhole::index_read_last");
+  THD *thd= ha_thd();
+  if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query == NULL)
+    DBUG_RETURN(0);
   DBUG_RETURN(HA_ERR_END_OF_FILE);
 }
 
diff --git a/storage/blackhole/ha_blackhole.h b/storage/blackhole/ha_blackhole.h
index d5a0d08926c..085840cce43 100644
--- a/storage/blackhole/ha_blackhole.h
+++ b/storage/blackhole/ha_blackhole.h
@@ -53,7 +53,7 @@ public:
   ulonglong table_flags() const
   {
     return(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
-           HA_BINLOG_STMT_CAPABLE |
+           HA_BINLOG_STMT_CAPABLE | HA_BINLOG_ROW_CAPABLE |
            HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
            HA_FILE_BASED | HA_CAN_GEOMETRY | HA_CAN_INSERT_DELAYED);
   }
@@ -72,7 +72,6 @@ public:
   uint max_supported_key_part_length() const { return BLACKHOLE_MAX_KEY_LENGTH; }
   int open(const char *name, int mode, uint test_if_locked);
   int close(void);
-  int write_row(uchar * buf);
   int rnd_init(bool scan);
   int rnd_next(uchar *buf);
   int rnd_pos(uchar * buf, uchar *pos);
@@ -94,4 +93,8 @@ public:
   THR_LOCK_DATA **store_lock(THD *thd,
                              THR_LOCK_DATA **to,
                              enum thr_lock_type lock_type);
+private:
+  virtual int write_row(uchar *buf);
+  virtual int update_row(const uchar *old_data, uchar *new_data);
+  virtual int delete_row(const uchar *buf);
 };