MDEV-31677 Assertion failed upon online ALTER with binlog_row_image=NOBLOB

Make binlog_prepare_row_images accept image type as an argument.

mysql-test/main/alter_table_online_debug.result

@@ -1439,6 +1439,24 @@ connection default;
 drop table t;
 set debug_sync= reset;
 set debug_dbug= @old_dbug;
+create table t (id int, a int, b text, primary key (id));
+insert into t values (1,10,''),(2,20,'');
+set debug_sync= 'alter_table_online_progress signal do_update wait_for go';
+alter table t force, algorithm=copy, lock=none;
+connection con1;
+set @old_binlog_row_image= @@binlog_row_image;
+set binlog_row_image= noblob;
+set debug_sync= 'now wait_for do_update';
+update t set a = 1;
+delete from t where id = 1;
+set debug_sync= 'now signal go';
+set binlog_row_image= @old_binlog_row_image;
+connection default;
+select * from t;
+id	a	b
+2	1	
+drop table t;
+set debug_sync= reset;
 disconnect con1;
 disconnect con2;
 #

mysql-test/main/alter_table_online_debug.test

@@ -1620,6 +1620,33 @@ while ($i) {
   dec $i;
 }
 
+
+
+
+create table t (id int, a int, b text, primary key (id));
+insert into t values (1,10,''),(2,20,'');
+
+set debug_sync= 'alter_table_online_progress signal do_update wait_for go';
+send alter table t force, algorithm=copy, lock=none;
+
+--connection con1
+set @old_binlog_row_image= @@binlog_row_image;
+set binlog_row_image= noblob;
+set debug_sync= 'now wait_for do_update';
+
+update t set a = 1;
+delete from t where id = 1;
+
+set debug_sync= 'now signal go';
+set binlog_row_image= @old_binlog_row_image;
+
+--connection default
+--reap
+select * from t;
+
+drop table t;
+set debug_sync= reset;
+
 --disconnect con1
 --disconnect con2
 --echo #

sql/handler.cc

@@ -7272,7 +7272,8 @@ static int binlog_log_row_to_binlog(TABLE* table,
                                      use_trans_cache(thd, has_trans));
 
     error= (*log_func)(thd, table, mysql_bin_log.as_event_log(), cache,
-                       has_trans, before_record, after_record);
+                       has_trans, thd->variables.binlog_row_image,
+                       before_record, after_record);
   DBUG_RETURN(error ? HA_ERR_RBR_LOGGING_FAILED : 0);
 }
 

sql/handler.h

@@ -659,7 +659,7 @@ class Cache_flip_event_log;
 class binlog_cache_data;
 class online_alter_cache_data;
 typedef bool Log_func(THD*, TABLE*, Event_log *, binlog_cache_data *, bool,
-                      const uchar*, const uchar*);
+                      ulong, const uchar*, const uchar*);
 
 /*
   These flags are set by the parser and describes the type of

sql/log.cc

@@ -2290,7 +2290,7 @@ int binlog_log_row_online_alter(TABLE* table, const uchar *before_record,
   table->rpl_write_set= &table->s->all_set;
 
   int error= (*log_func)(thd, table, table->s->online_alter_binlog,
-                         table->online_alter_cache, true,
+                         table->online_alter_cache, true, BINLOG_ROW_IMAGE_FULL,
                          before_record, after_record);
 
   table->rpl_write_set= old_rpl_write_set;

sql/log_event.h

@@ -4922,7 +4922,7 @@ class Write_rows_log_event : public Rows_log_event
   static bool binlog_row_logging_function(THD *thd, TABLE *table,
                                           Event_log *bin_log,
                                           binlog_cache_data *cache_data,
-                                          bool is_transactional,
+                                          bool is_transactional, ulong,
                                           const uchar *before_record
                                           __attribute__((unused)),
                                           const uchar *after_record)
@@ -5006,11 +5006,13 @@ class Update_rows_log_event : public Rows_log_event
                                           Event_log *bin_log,
                                           binlog_cache_data *cache_data,
                                           bool is_transactional,
+                                          ulong row_image,
                                           const uchar *before_record,
                                           const uchar *after_record)
   {
     DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
     return thd->binlog_update_row(table, bin_log, cache_data, is_transactional,
+                                  (enum_binlog_row_image)row_image,
                                   before_record, after_record);
   }
 #endif
@@ -5094,12 +5096,14 @@ class Delete_rows_log_event : public Rows_log_event
                                           Event_log *bin_log,
                                           binlog_cache_data *cache_data,
                                           bool is_transactional,
+                                          ulong row_image,
                                           const uchar *before_record,
                                           const uchar *after_record
                                           __attribute__((unused)))
   {
     DBUG_ASSERT(!table->versioned(VERS_TRX_ID));
     return thd->binlog_delete_row(table, bin_log, cache_data, is_transactional,
+                                  (enum_binlog_row_image)row_image,
                                   before_record);
   }
 #endif

sql/sql_class.cc

@@ -7053,6 +7053,7 @@ int THD::binlog_write_row(TABLE* table, Event_log *bin_log,
 
 int THD::binlog_update_row(TABLE* table,  Event_log *bin_log,
                            binlog_cache_data *cache_data, bool is_trans,
+                           enum_binlog_row_image row_image,
                            const uchar *before_record,
                            const uchar *after_record)
 {
@@ -7070,7 +7071,7 @@ int THD::binlog_update_row(TABLE* table,  Event_log *bin_log,
      not needed for binlogging. This is done according to the:
      binlog-row-image option.
    */
-  binlog_prepare_row_images(table);
+  binlog_prepare_row_images(table, row_image);
 
   size_t const before_maxlen= max_row_length(table, table->read_set,
                                              before_record);
@@ -7122,6 +7123,7 @@ int THD::binlog_update_row(TABLE* table,  Event_log *bin_log,
 
 int THD::binlog_delete_row(TABLE* table, Event_log *bin_log,
                            binlog_cache_data *cache_data, bool is_trans,
+                           enum_binlog_row_image row_image,
                            uchar const *record)
 {
   /**
@@ -7138,7 +7140,7 @@ int THD::binlog_delete_row(TABLE* table, Event_log *bin_log,
      not needed for binlogging. This is done according to the:
      binlog-row-image option.
    */
-  binlog_prepare_row_images(table);
+  binlog_prepare_row_images(table, row_image);
 
   /*
      Pack records into format for transfer. We are allocating more
@@ -7179,8 +7181,7 @@ int THD::binlog_delete_row(TABLE* table, Event_log *bin_log,
    Remove from read_set spurious columns. The write_set has been
    handled before in table->mark_columns_needed_for_update.
 */
-
-void THD::binlog_prepare_row_images(TABLE *table)
+void binlog_prepare_row_images(TABLE *table, enum_binlog_row_image row_image)
 {
   DBUG_ENTER("THD::binlog_prepare_row_images");
 
@@ -7194,7 +7195,7 @@ void THD::binlog_prepare_row_images(TABLE *table)
     and the handler involved supports this.
    */
   if (table->s->primary_key < MAX_KEY &&
-      (thd->variables.binlog_row_image < BINLOG_ROW_IMAGE_FULL) &&
+      row_image < BINLOG_ROW_IMAGE_FULL &&
       !ha_check_storage_engine_flag(table->s->db_type(),
                                     HTON_NO_BINLOG_ROW_OPT))
   {

sql/sql_class.h

@@ -2960,15 +2960,15 @@ class THD: public THD_count, /* this must be first */
                        const uchar *buf);
   int binlog_delete_row(TABLE* table,  Event_log *bin_log,
                         binlog_cache_data *cache_data, bool is_transactional,
-                        const uchar *buf);
+                        enum_binlog_row_image row_image, const uchar *buf);
   int binlog_update_row(TABLE* table, Event_log *bin_log,
                         binlog_cache_data *cache_data, bool is_transactional,
+                        enum_binlog_row_image row_image,
                         const uchar *old_data, const uchar *new_data);
   bool prepare_handlers_for_update(uint flag);
   bool binlog_write_annotated_row(Log_event_writer *writer);
   void binlog_prepare_for_row_logging();
   bool binlog_write_table_maps();
-  static void binlog_prepare_row_images(TABLE* table);
 
   void set_server_id(uint32 sid) { variables.server_id = sid; }
 
@@ -7641,6 +7641,8 @@ inline bool binlog_should_compress(size_t len)
     len >= opt_bin_log_compress_min_len;
 }
 
+void binlog_prepare_row_images(TABLE* table,
+                               enum_binlog_row_image row_image);
 
 /**
    Save thd sql_mode on instantiation.