SQL: UPDATE on row-based replication [closes: #94]

mysql-test/suite/versioning/r/rpl_row.result

+include/master-slave.inc
+[connection master]
+connection slave;
+connection master;
+CREATE TABLE t1 (x int) with system versioning ENGINE = innodb;
+insert into t1 values (1);
+SELECT * FROM t1;
+x
+1
+delete from t1;
+select * from t1;
+x
+select * from t1 for system_time all;
+x
+1
+connection slave;
+select * from t1;
+x
+select * from t1 for system_time all;
+x
+1
+connection master;
+insert into t1 values (2);
+connection slave;
+select * from t1;
+x
+2
+connection master;
+update t1 set x = 3;
+connection slave;
+select * from t1;
+x
+3
+select * from t1 for system_time all;
+x
+1
+3
+2
+connection master;
+drop table t1;
+include/rpl_end.inc

mysql-test/suite/versioning/r/rpl_stmt.result

+include/master-slave.inc
+[connection master]
+connection slave;
+connection master;
+CREATE TABLE t1 (x int) with system versioning ENGINE = innodb;
+insert into t1 values (1);
+SELECT * FROM t1;
+x
+1
+delete from t1;
+select * from t1;
+x
+select * from t1 for system_time all;
+x
+1
+connection slave;
+select * from t1;
+x
+select * from t1 for system_time all;
+x
+1
+connection master;
+insert into t1 values (2);
+connection slave;
+select * from t1;
+x
+2
+connection master;
+update t1 set x = 3;
+connection slave;
+select * from t1;
+x
+3
+select * from t1 for system_time all;
+x
+1
+3
+2
+connection master;
+drop table t1;
+include/rpl_end.inc

mysql-test/suite/versioning/t/rpl_row.test

+-- source include/have_binlog_format_row.inc
+-- source include/master-slave.inc
+-- source include/have_innodb.inc
+
+-- source rpl_test.inc
+
+-- source include/rpl_end.inc

mysql-test/suite/versioning/t/rpl_stmt.test

+-- source include/have_binlog_format_statement.inc
+-- source include/master-slave.inc
+-- source include/have_innodb.inc
+
+-- source rpl_test.inc
+
+-- source include/rpl_end.inc

mysql-test/suite/versioning/t/rpl_test.inc

+#BUG#12662190 - COM_COMMIT IS NOT INCREMENTED FROM THE BINARY LOGS ON SLAVE, COM_BEGIN IS
+#Testing command counters -BEFORE.
+#Storing the before counts of Slave
+connection slave;
+let $slave_com_commit_before= query_get_value(SHOW GLOBAL STATUS LIKE 'com_commit', Value, 1);
+let $slave_com_insert_before= query_get_value(SHOW GLOBAL STATUS LIKE 'com_insert', Value, 1);
+let $slave_com_delete_before= query_get_value(SHOW GLOBAL STATUS LIKE 'com_delete', Value, 1);
+let $slave_com_update_before= query_get_value(SHOW GLOBAL STATUS LIKE 'com_update', Value, 1);
+
+connection master;
+CREATE TABLE t1 (x int) with system versioning ENGINE = innodb;
+insert into t1 values (1);
+SELECT * FROM t1;
+delete from t1;
+select * from t1;
+select * from t1 for system_time all;
+sync_slave_with_master;
+select * from t1;
+select * from t1 for system_time all;
+
+connection master;
+insert into t1 values (2);
+sync_slave_with_master;
+select * from t1;
+
+connection master;
+update t1 set x = 3;
+sync_slave_with_master;
+select * from t1;
+select * from t1 for system_time all;
+
+connection master;
+drop table t1;

sql/ha_sequence.cc

@@ -220,6 +220,7 @@ int ha_sequence::write_row(uchar *buf)
       sequence->copy(&tmp_seq);
     rows_changed++;
     /* We have to do the logging while we hold the sequence mutex */
+    if (table->file->check_table_binlog_row_based(1))
       error= binlog_log_row(table, 0, buf, log_func);
     row_already_logged= 1;
   }
@@ -254,6 +255,7 @@ int ha_sequence::update_row(const uchar *old_data, uchar *new_data)
     sequence->copy(&tmp_seq);
     rows_changed++;
     /* We have to do the logging while we hold the sequence mutex */
+    if (table->file->check_table_binlog_row_based(1))
       error= binlog_log_row(table, old_data, new_data,
                           Update_rows_log_event::binlog_row_logging_function);
     row_already_logged= 1;

sql/handler.cc

@@ -5795,7 +5795,7 @@ static int write_locked_table_maps(THD *thd)
 
 static int check_wsrep_max_ws_rows();
 
-static int binlog_log_row_internal(TABLE* table,
+int binlog_log_row(TABLE* table,
                           const uchar *before_record,
                           const uchar *after_record,
                           Log_func *log_func)
@@ -5833,16 +5833,6 @@ static int binlog_log_row_internal(TABLE* table,
   return error ? HA_ERR_RBR_LOGGING_FAILED : 0;
 }
 
-int binlog_log_row(TABLE* table,
-                   const uchar *before_record,
-                   const uchar *after_record,
-                   Log_func *log_func)
-{
-  if (!table->file->check_table_binlog_row_based(1))
-    return 0;
-  return binlog_log_row_internal(table, before_record, after_record, log_func);
-}
-
 
 int handler::ha_external_lock(THD *thd, int lock_type)
 {
@@ -5991,6 +5981,7 @@ int handler::ha_write_row(uchar *buf)
   if (likely(!error) && !row_already_logged)
   {
     rows_changed++;
+    if (table->file->check_table_binlog_row_based(1))
       error= binlog_log_row(table, 0, buf, log_func);
   }
   DEBUG_SYNC_C("ha_write_row_end");
@@ -6012,6 +6003,9 @@ int handler::ha_update_row(const uchar *old_data, uchar *new_data)
   DBUG_ASSERT(new_data == table->record[0]);
   DBUG_ASSERT(old_data == table->record[1]);
 
+  // InnoDB changes sys_trx_end to curr_trx_id and we need to restore MAX_TRX
+  if (table->file->check_table_binlog_row_based(1))
+    memcpy(table->record[2], table->record[1], table->s->reclength);
   MYSQL_UPDATE_ROW_START(table_share->db.str, table_share->table_name.str);
   mark_trx_read_write();
   increment_statistics(&SSV::ha_update_count);
@@ -6023,8 +6017,11 @@ int handler::ha_update_row(const uchar *old_data, uchar *new_data)
   if (likely(!error) && !row_already_logged)
   {
     rows_changed++;
+    if (table->file->check_table_binlog_row_based(1)) {
+      memcpy(table->record[1], table->record[2], table->s->reclength);
       error= binlog_log_row(table, old_data, new_data, log_func);
     }
+  }
   return error;
 }
 
@@ -6072,6 +6069,7 @@ int handler::ha_delete_row(const uchar *buf)
   if (likely(!error))
   {
     rows_changed++;
+    if (table->file->check_table_binlog_row_based(1))
       error= binlog_log_row(table, buf, 0, log_func);
   }
   return error;