MDEV-31776 Online ALTER reports the number of affected rows incorrectly

Add a new virtual function that will increase the inserted rows count
for the insert log event and decrease it for the delete event.

Reuses Rows_log_event::m_row_count on the replication side, which was only
set on the logging side.

mysql-test/main/alter_table_online_debug.result

@@ -16,6 +16,8 @@ connection con2;
 insert into t1 values (123), (456), (789);
 set debug_sync= 'now SIGNAL end';
 connection default;
+affected rows: 4
+info: Records: 4  Duplicates: 0  Warnings: 0
 select * from t1;
 a	b
 5	NULL
@@ -123,6 +125,8 @@ connection con2;
 update t1 set b= 55 where a = 1;
 set debug_sync= 'now SIGNAL end';
 connection default;
+affected rows: 2
+info: Records: 2  Duplicates: 0  Warnings: 0
 select * from t1;
 a	b	c
 1	55	1

mysql-test/main/alter_table_online_debug.test

@@ -33,7 +33,9 @@ insert into t1 values (123), (456), (789);
 set debug_sync= 'now SIGNAL end';
 
 --connection default
+--enable_info
 --reap
+--disable_info
 select * from t1;
 
 --echo # Insert, error
@@ -167,7 +169,9 @@ update t1 set b= 55 where a = 1;
 set debug_sync= 'now SIGNAL end';
 
 --connection default
+--enable_info
 --reap
+--disable_info
 select * from t1;
 
 --echo # Update and add a column in the middle

sql/log_event.h

@@ -1564,6 +1564,12 @@ class Log_event
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
 
+  /**
+    Increase or decrease the rows inserted during ALTER TABLE based on the event
+    type.
+  */
+  virtual void online_alter_update_row_count(ha_rows *) const {}
+
   /**
      Apply the event to the database.
 
@@ -4973,6 +4979,11 @@ class Write_rows_log_event : public Rows_log_event
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
   uint8 get_trg_event_map() const override;
+
+  void online_alter_update_row_count(ha_rows *rows) const override
+  {
+    *rows += m_row_count;
+  }
 #endif
 
 private:
@@ -5148,6 +5159,11 @@ class Delete_rows_log_event : public Rows_log_event
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
   uint8 get_trg_event_map() const override;
+
+  void online_alter_update_row_count(ha_rows *rows) const override
+  {
+    *rows -= m_row_count;
+  }
 #endif
 
 protected:

sql/log_event_server.cc

@@ -5184,6 +5184,7 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
           thd->transaction->stmt.modified_non_trans_table= TRUE;
       if (likely(error == 0))
       {
+        m_row_count++;
         error= thd->killed_errno();
         if (error && !thd->is_error())
           my_error(error, MYF(0));

sql/sql_table.cc

@@ -11656,7 +11656,8 @@ class Has_default_error_handler : public Internal_error_handler
 
 
 static int online_alter_read_from_binlog(THD *thd, rpl_group_info *rgi,
-                                         Cache_flip_event_log *log)
+                                         Cache_flip_event_log *log,
+                                         ha_rows *found_rows)
 {
   int error= 0;
 
@@ -11681,8 +11682,11 @@ static int online_alter_read_from_binlog(THD *thd, rpl_group_info *rgi,
 
     ev->thd= thd;
     error= ev->apply_event(rgi);
-    if (thd->is_error())
-      error= 1;
+
+    error= error || thd->is_error();
+    if(likely(!error))
+      ev->online_alter_update_row_count(found_rows);
+
     if (ev != rgi->rli->relay_log.description_event_for_exec)
       delete ev;
     thd_progress_report(thd, my_b_tell(log_file), thd->progress.max_counter);
@@ -12108,7 +12112,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     mysql_unlock_tables(thd, thd->lock);
     thd->lock= NULL;
 
-    error= online_alter_read_from_binlog(thd, &rgi, binlog);
+    error= online_alter_read_from_binlog(thd, &rgi, binlog, &found_count);
     if (start_alter_id)
     {
       DBUG_ASSERT(thd->slave_thread);
@@ -12129,7 +12133,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     if (!error)
     {
       thd_progress_next_stage(thd);
-      error= online_alter_read_from_binlog(thd, &rgi, binlog);
+      error= online_alter_read_from_binlog(thd, &rgi, binlog, &found_count);
     }
     if (error)
       from->s->tdc->flush_unused(1); // to free the binlog