MDEV-8203 Assert in Query_log_event::do_apply_event()

This happens because the master writes a table_map event to the binary log, but no row event.
The slave has a check that there should always be a row event if there was a table_map event, which
causes a crash.

Fixed by remembering in the cache what kind of events are logged
and ignore cached statements which is just a table map event.

mysql-test/suite/rpl/r/rpl_trans_no_trans.result

+include/master-slave.inc
+[connection master]
+create or replace table t1(id int)engine=innodb;
+create or replace table t3(id int)engine=myisam;
+create or replace function t99 (a int)
+returns int(10)
+MODIFIES SQL DATA
+begin
+if (a > 100)
+then
+insert into t3 values (a);
+end if;
+return a;
+end//
+begin;
+insert into t1 values(t99(1));
+insert into t1 values(t99(101));
+commit;
+select * from t1;
+id
+1
+101
+select * from t3;
+id
+101
+insert into t1 values(t99(1));
+drop function t99;
+drop table t1,t3;
+connection slave;
+connection master;
+CREATE TABLE t1 (i INT) ENGINE=InnoDB;
+CREATE TABLE t2 (j INT) ENGINE=MyISAM;
+CREATE TRIGGER tr AFTER INSERT ON t1 FOR EACH ROW 
+BEGIN 
+SET @a = unknown_column_just_to_raise_an_error; 
+INSERT INTO t2 VALUES (NULL) ; 
+END||
+INSERT INTO t1 VALUES (1);
+ERROR 42S22: Unknown column 'unknown_column_just_to_raise_an_error' in 'field list'
+connection slave;
+connection master;
+drop trigger tr;
+drop table t1,t2;
+include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_trans_no_trans.test

+#
+# Test mixing transactional and non transactional tables
+#
+
+--source include/master-slave.inc
+--source include/have_innodb.inc
+--source include/have_binlog_format_row.inc
+
+#
+# Test updating conditionally a non transactinal table in a function
+# This verifies that we don't write only a table map for a non transactional,
+# without any row events
+# The original bug caused a crash on the slave when doing a sync_slave
+#
+
+create or replace table t1(id int)engine=innodb;
+create or replace table t3(id int)engine=myisam;
+
+delimiter //;
+create or replace function t99 (a int)
+returns int(10)
+MODIFIES SQL DATA
+begin
+  if (a > 100)
+  then
+    insert into t3 values (a);
+  end if;
+  return a;
+end//
+delimiter ;//
+begin;
+insert into t1 values(t99(1));
+insert into t1 values(t99(101));
+commit;
+select * from t1;
+select * from t3;
+insert into t1 values(t99(1));
+
+drop function t99;
+drop table t1,t3;
+
+sync_slave_with_master;
+connection master;
+
+#
+# MDEV-8203
+# Assertion `!current_stmt_is_commit || !rgi->tables_to_lock' failed in
+# Query_log_event::do_apply_event(rpl_group_info*, const char*, uint32)
+#
+
+CREATE TABLE t1 (i INT) ENGINE=InnoDB;
+CREATE TABLE t2 (j INT) ENGINE=MyISAM;
+
+--delimiter ||
+CREATE TRIGGER tr AFTER INSERT ON t1 FOR EACH ROW 
+BEGIN 
+  SET @a = unknown_column_just_to_raise_an_error; 
+  INSERT INTO t2 VALUES (NULL) ; 
+END||
+--delimiter ;
+
+--error ER_BAD_FIELD_ERROR
+INSERT INTO t1 VALUES (1);
+--sync_slave_with_master
+
+connection master;
+
+drop trigger tr;
+drop table t1,t2;
+
+# End of 4.1 tests
+--source include/rpl_end.inc

sql/log.cc

@@ -250,7 +250,8 @@ void make_default_log_name(char **out, const char* log_ext, bool once)
 class binlog_cache_data
 {
 public:
-  binlog_cache_data(): m_pending(0), before_stmt_pos(MY_OFF_T_UNDEF),
+  binlog_cache_data(): m_pending(0), status(0),
+  before_stmt_pos(MY_OFF_T_UNDEF),
   incident(FALSE), changes_to_non_trans_temp_table_flag(FALSE),
   saved_max_binlog_cache_size(0), ptr_binlog_cache_use(0),
   ptr_binlog_cache_disk_use(0)
@@ -262,9 +263,22 @@ class binlog_cache_data
     close_cached_file(&cache_log);
   }
 
+  /*
+    Return 1 if there is no relevant entries in the cache
+
+    This is:
+    - Cache is empty
+    - There are row or critical (DDL?) events in the cache
+
+    The status test is needed to avoid writing entries with only
+    a table map entry, which would crash in do_apply_event() on the slave
+    as it assumes that there is always a row entry after a table map.
+  */
   bool empty() const
   {
-    return pending() == NULL && my_b_tell(&cache_log) == 0;
+    return (pending() == NULL &&
+            (my_b_write_tell(&cache_log) == 0 ||
+             ((status & (LOGGED_ROW_EVENT | LOGGED_CRITICAL)) == 0)));
   }
 
   Rows_log_event *pending() const
@@ -305,6 +319,7 @@ class binlog_cache_data
       my_chsize(cache_log.file, 0, 0, MYF(MY_WME));
 
     changes_to_non_trans_temp_table_flag= FALSE;
+    status= 0;
     incident= FALSE;
     before_stmt_pos= MY_OFF_T_UNDEF;
     /*
@@ -376,6 +391,11 @@ class binlog_cache_data
     cache_log.end_of_file= saved_max_binlog_cache_size;
   }
 
+  void add_status(enum_logged_status status_arg)
+  {
+    status|= status_arg;
+  }
+
   /*
     Cache to store data before copying it to the binary log.
   */
@@ -388,6 +408,13 @@ class binlog_cache_data
    */
   Rows_log_event *m_pending;
 
+  /*
+    Bit flags for what has been writting to cache. Used to
+    discard logs without any data changes.
+    see enum_logged_status;
+  */
+  uint32 status;
+
   /*
     Binlog position before the start of the current statement.
   */
@@ -459,6 +486,13 @@ class binlog_cache_data
   binlog_cache_data(const binlog_cache_data& info);
 };
 
+
+void Log_event_writer::add_status(enum_logged_status status)
+{
+  if (likely(cache_data))
+    cache_data->add_status(status);
+}
+
 class binlog_cache_mngr {
 public:
   binlog_cache_mngr(my_off_t param_max_binlog_stmt_cache_size,
@@ -5207,12 +5241,14 @@ int MYSQL_BIN_LOG::new_file_impl(bool need_lock)
   DBUG_RETURN(error);
 }
 
-bool MYSQL_BIN_LOG::write_event(Log_event *ev, IO_CACHE *file)
+bool MYSQL_BIN_LOG::write_event(Log_event *ev, binlog_cache_data *cache_data,
+                                IO_CACHE *file)
 {
-  Log_event_writer writer(file, &crypto);
+  Log_event_writer writer(file, 0, &crypto);
   if (crypto.scheme && file == &log_file)
     writer.ctx= alloca(crypto.ctx_size);
-
+  if (cache_data)
+    cache_data->add_status(ev->logged_status());
   return writer.write(ev);
 }
 
@@ -5649,10 +5685,11 @@ int THD::binlog_write_table_map(TABLE *table, bool is_transactional,
 
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(this, binlog_hton);
+  binlog_cache_data *cache_data= (cache_mngr->
+                                  get_binlog_cache_data(is_transactional));
+  IO_CACHE *file= &cache_data->cache_log;
+  Log_event_writer writer(file, cache_data);
 
-  IO_CACHE *file=
-    cache_mngr->get_binlog_cache_log(use_trans_cache(this, is_transactional));
-  Log_event_writer writer(file);
   if (with_annotate && *with_annotate)
   {
     Annotate_rows_log_event anno(table->in_use, is_transactional, false);
@@ -5788,7 +5825,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
 
   if (Rows_log_event* pending= cache_data->pending())
   {
-    Log_event_writer writer(&cache_data->cache_log);
+    Log_event_writer writer(&cache_data->cache_log, cache_data);
 
     /*
       Write pending event to the cache.
@@ -6200,8 +6237,8 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
         goto err;
 
       is_trans_cache= use_trans_cache(thd, using_trans);
-      file= cache_mngr->get_binlog_cache_log(is_trans_cache);
       cache_data= cache_mngr->get_binlog_cache_data(is_trans_cache);
+      file= &cache_data->cache_log;
 
       if (thd->lex->stmt_accessed_non_trans_temp_table())
         cache_data->set_changes_to_non_trans_temp_table();
@@ -6225,21 +6262,19 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
       Annotate_rows_log_event anno(thd, using_trans, direct);
       /* Annotate event should be written not more than once */
       *with_annotate= 0;
-      if (write_event(&anno, file))
+      if (write_event(&anno, cache_data, file))
         goto err;
     }
 
-    if (thd)
     {
       if (!thd->is_current_stmt_binlog_format_row())
       {
-
         if (thd->stmt_depends_on_first_successful_insert_id_in_prev_stmt)
         {
           Intvar_log_event e(thd,(uchar) LAST_INSERT_ID_EVENT,
                              thd->first_successful_insert_id_in_prev_stmt_for_binlog,
                              using_trans, direct);
-          if (write_event(&e, file))
+          if (write_event(&e, cache_data, file))
             goto err;
         }
         if (thd->auto_inc_intervals_in_cur_stmt_for_binlog.nb_elements() > 0)
@@ -6250,14 +6285,14 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
           Intvar_log_event e(thd, (uchar) INSERT_ID_EVENT,
                              thd->auto_inc_intervals_in_cur_stmt_for_binlog.
                              minimum(), using_trans, direct);
-          if (write_event(&e, file))
+          if (write_event(&e, cache_data, file))
             goto err;
         }
         if (thd->rand_used)
         {
           Rand_log_event e(thd,thd->rand_saved_seed1,thd->rand_saved_seed2,
                            using_trans, direct);
-          if (write_event(&e, file))
+          if (write_event(&e, cache_data, file))
             goto err;
         }
         if (thd->user_var_events.elements)
@@ -6281,7 +6316,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
                                  flags,
                                  using_trans,
                                  direct);
-            if (write_event(&e, file))
+            if (write_event(&e, cache_data, file))
               goto err;
           }
         }
@@ -6291,7 +6326,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
     /*
       Write the event.
     */
-    if (write_event(event_info, file) ||
+    if (write_event(event_info, cache_data, file) ||
         DBUG_EVALUATE_IF("injecting_fault_writing", 1, 0))
       goto err;
 
@@ -6676,7 +6711,8 @@ class CacheWriter: public Log_event_writer
 
   CacheWriter(THD *thd_arg, IO_CACHE *file_arg, bool do_checksum,
               Binlog_crypt_data *cr)
-    : Log_event_writer(file_arg, cr), remains(0), thd(thd_arg), first(true)
+    : Log_event_writer(file_arg, 0, cr), remains(0), thd(thd_arg),
+      first(true)
   { checksum_len= do_checksum ? BINLOG_CHECKSUM_LEN : 0; }
 
   ~CacheWriter()
@@ -6684,6 +6720,7 @@ class CacheWriter: public Log_event_writer
 
   int write(uchar* pos, size_t len)
   {
+    DBUG_ENTER("CacheWriter::write");
     if (first)
       write_header(pos, len);
     else
@@ -6692,7 +6729,7 @@ class CacheWriter: public Log_event_writer
     remains -= len;
     if ((first= !remains))
       write_footer();
-    return 0;
+    DBUG_RETURN(0);
   }
 private:
   THD *thd;

sql/log.h

@@ -415,8 +415,10 @@ class MYSQL_QUERY_LOG: public MYSQL_LOG
   ( ((ulong)(c)>>1) == BINLOG_COOKIE_DUMMY_ID )
 
 class binlog_cache_mngr;
+class binlog_cache_data;
 struct rpl_gtid;
 struct wait_for_commit;
+
 class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
 {
  private:
@@ -743,8 +745,8 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   void stop_union_events(THD *thd);
   bool is_query_in_union(THD *thd, query_id_t query_id_param);
 
-  bool write_event(Log_event *ev, IO_CACHE *file);
-  bool write_event(Log_event *ev) { return write_event(ev, &log_file); }
+  bool write_event(Log_event *ev, binlog_cache_data *data, IO_CACHE *file);
+  bool write_event(Log_event *ev) { return write_event(ev, 0, &log_file); }
 
   bool write_event_buffer(uchar* buf,uint len);
   bool append(Log_event* ev);

sql/log_event.cc

@@ -9292,7 +9292,7 @@ int Create_file_log_event::do_apply_event(rpl_group_info *rgi)
   char *ext;
   int fd = -1;
   IO_CACHE file;
-  Log_event_writer lew(&file);
+  Log_event_writer lew(&file, 0);
   int error = 1;
   Relay_log_info const *rli= rgi->rli;
 

sql/log_event.h

@@ -716,6 +716,21 @@ enum Log_event_type
   ENUM_END_EVENT /* end marker */
 };
 
+
+/*
+  Bit flags for what has been writting to cache. Used to
+  discard logs with table map events but not row events and
+  nothing else important. This is stored by cache.
+*/
+
+enum enum_logged_status
+{
+  LOGGED_TABLE_MAP= 1,
+  LOGGED_ROW_EVENT= 2,
+  LOGGED_NO_DATA=   4,
+  LOGGED_CRITICAL=  8
+};
+
 static inline bool LOG_EVENT_IS_QUERY(enum Log_event_type type)
 {
   return type == QUERY_EVENT || type == QUERY_COMPRESSED_EVENT;
@@ -785,6 +800,7 @@ class THD;
 
 class Format_description_log_event;
 class Relay_log_info;
+class binlog_cache_data;
 
 #ifdef MYSQL_CLIENT
 enum enum_base64_output_mode {
@@ -896,6 +912,7 @@ typedef struct st_print_event_info
   This class encapsulates writing of Log_event objects to IO_CACHE.
   Automatically calculates the checksum and encrypts the data, if necessary.
 */
+
 class Log_event_writer
 {
 public:
@@ -907,13 +924,16 @@ class Log_event_writer
   int write_data(const uchar *pos, size_t len);
   int write_footer();
   my_off_t pos() { return my_b_safe_tell(file); }
+  void add_status(enum_logged_status status);
 
-Log_event_writer(IO_CACHE *file_arg, Binlog_crypt_data *cr= 0)
+  Log_event_writer(IO_CACHE *file_arg, binlog_cache_data *cache_data_arg,
+                   Binlog_crypt_data *cr= 0)
   : bytes_written(0), ctx(0),
-    file(file_arg), crypto(cr) { }
+    file(file_arg), cache_data(cache_data_arg), crypto(cr) { }
 
 private:
   IO_CACHE *file;
+  binlog_cache_data *cache_data;
   /**
     Placeholder for event checksum while writing to binlog.
    */
@@ -1236,7 +1256,7 @@ class Log_event
                     bool is_more);
   void print_base64(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info,
                     bool is_more);
-#endif
+#endif /* MYSQL_SERVER */
 
   /* The following code used for Flashback */
 #ifdef MYSQL_CLIENT
@@ -1382,6 +1402,7 @@ class Log_event
   }
 #endif
   virtual Log_event_type get_type_code() = 0;
+  virtual enum_logged_status logged_status() { return LOGGED_CRITICAL; }
   virtual bool is_valid() const = 0;
   virtual my_off_t get_header_len(my_off_t len) { return len; }
   void set_artificial_event() { flags |= LOG_EVENT_ARTIFICIAL_F; }
@@ -3361,6 +3382,7 @@ class Gtid_log_event: public Log_event
                  const Format_description_log_event *description_event);
   ~Gtid_log_event() { }
   Log_event_type get_type_code() { return GTID_EVENT; }
+  enum_logged_status logged_status() { return LOGGED_NO_DATA; }
   int get_data_size()
   {
     return GTID_HEADER_LEN + ((flags2 & FL_GROUP_COMMIT_ID) ? 2 : 0);
@@ -3857,6 +3879,7 @@ class Annotate_rows_log_event: public Log_event
 
   virtual int get_data_size();
   virtual Log_event_type get_type_code();
+  enum_logged_status logged_status() { return LOGGED_NO_DATA; }
   virtual bool is_valid() const;
   virtual bool is_part_of_group() { return 1; }
 
@@ -4271,6 +4294,7 @@ class Table_map_log_event : public Log_event
   const char *get_db_name() const    { return m_dbnam; }
 
   virtual Log_event_type get_type_code() { return TABLE_MAP_EVENT; }
+  virtual enum_logged_status logged_status() { return LOGGED_TABLE_MAP; }
   virtual bool is_valid() const { return m_memory != NULL; /* we check malloc */ }
   virtual bool is_part_of_group() { return 1; }
 
@@ -4398,6 +4422,7 @@ class Rows_log_event : public Log_event
   flag_set get_flags(flag_set flags_arg) const { return m_flags & flags_arg; }
 
   Log_event_type get_type_code() { return m_type; } /* Specific type (_V1 etc) */
+  enum_logged_status logged_status() { return LOGGED_ROW_EVENT; }
   virtual Log_event_type get_general_type_code() = 0; /* General rows op type, no version */
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
@@ -5160,6 +5185,7 @@ inline int Log_event_writer::write(Log_event *ev)
   ev->writer= this;
   int res= ev->write();
   IF_DBUG(ev->writer= 0,); // writer must be set before every Log_event::write
+  add_status(ev->logged_status());
   return res;
 }
 

sql/wsrep_binlog.cc

@@ -447,7 +447,7 @@ void wsrep_dump_rbr_buf_with_header(THD *thd, const void *rbr_buf,
   char filename[PATH_MAX]= {0};
   File file;
   IO_CACHE cache;
-  Log_event_writer writer(&cache);
+  Log_event_writer writer(&cache, 0);
   Format_description_log_event *ev;
 
   int len= my_snprintf(filename, PATH_MAX, "%s/GRA_%lld_%lld_v2.log",

sql/wsrep_mysqld.cc

@@ -1245,7 +1245,7 @@ int wsrep_to_buf_helper(
     THD* thd, const char *query, uint query_len, uchar** buf, size_t* buf_len)
 {
   IO_CACHE tmp_io_cache;
-  Log_event_writer writer(&tmp_io_cache);
+  Log_event_writer writer(&tmp_io_cache,0);
   if (open_cached_file(&tmp_io_cache, mysql_tmpdir, TEMP_PREFIX,
                        65536, MYF(MY_WME)))
     return 1;