MDEV-15990 Refactor write_record and fix idempotent replication

See also MDEV-30046.

Idempotent write_row works same as REPLACE: if there is a duplicating
record in the table, then it will be deleted and re-inserted, with the
same update optimization.

The code in Rows:log_event::write_row was basically copy-pasted from
write_record.

What's done:
REPLACE operation was unified across replication and sql. It is now
representred as a Write_record class, that holds the whole state, and allows
re-using some resources in between the row writes.

Replace, IODKU and single insert implementations are split across different
methods, reluting in a much cleaner code.

The entry point is preserved as a single Write_record::write_record() call.
The implementation to call is chosen on the constructor stage.

This allowed several optimizations to be done:
1. The table key list is not iterated for every row. We find last unique key in
the order of checking once and preserve it across the rows. See last_uniq_key().
2. ib_handler::referenced_by_foreign_key acquires a global lock. This call was
done per row as well. Not all the table config that allows optimized replace is
folded into a single boolean field can_optimize. All the fields to check are
even stored in a single register on a 64-bit platform.
3. DUP_REPLACE and DUP_UPDATE cases now have one less level of indirection
4. modified_non_trans_tables is checked and set only when it's really needed.
5. Obsolete bitmap manipulations are removed.

Also:
* Unify replace initialization step across implementations:
  add prepare_for_replace and finalize_replace
* alloca is removed in favor of mem_root allocation. This memory is reused
  across the rows.
* An rpl-related callback is added to the replace branch, meaning that an extra
check is made per row replace even for the common case. It can be avoided with
templates if considered a problem.

Author: FooBarrior

mysql-test/main/long_unique_bugs_replication.result

@@ -9,20 +9,64 @@ insert into t1 values (2,2);
 update t1 set a1 = 'd' limit 1;
 update t1 set a1 = 'd2' where i1= 2;
 connection slave;
+connection slave;
+select * from t1;
+i1	a1
+1	d
+2	d2
 connection master;
 drop table t1;
+connection slave;
+connection master;
 #
 # MDEV-32093 long uniques break old->new replication
 #
-connection slave;
 create table t1 (id int not null, b1 varchar(255) not null, b2 varchar(2550) not null, unique (id), unique key (b1,b2) using hash) default charset utf8mb3;
 set global slave_exec_mode=idempotent;
 binlog 'aRf2ZA8BAAAA/AAAAAABAAAAAAQAMTAuNS4xNS1NYXJpYURCLWxvZwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABpF/ZkEzgNAAgAEgAEBAQEEgAA5AAEGggAAAAICAgCAAAACgoKAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEEwQADQgICAoKCgFRmTlk';
 binlog 'bBf2ZBMBAAAANAAAAJgHAAAAAHEAAAAAAAEABHRlc3QAAnQxAAQDDw8IBP0C4h0AeqMD4A==bBf2ZBcBAAAANAAAAMwHAAAAAHEAAAAAAAEABP/wj6QAAAEAYgEAZa6/VU0JAAAANteqUw==';
 binlog 'bBf2ZBMBAAAANAAAAJgHAAAAAHEAAAAAAAEABHRlc3QAAnQxAAQDDw8IBP0C4h0AeqMD4A==bBf2ZBcBAAAANAAAAMwHAAAAAHEAAAAAAAEABP/wj6QAAAEAYgEAZa6/VU0JAAAANteqUw==';
 binlog 'bBf2ZBMBAAAANAAAAHUkAAAAAHEAAAAAAAEABHRlc3QAAnQxAAQDDw8IBP0C4h0AaTGFIg==bBf2ZBgBAAAASAAAAL0kAAAAAHEAAAAAAAEABP//8I+kAAABAGIBAGWuv1VNCQAAAPBuWwAAAQBiAQBlrr9VTQkAAADxS9Lu';
-drop table t1;
+connection slave;
+select * from t1;
+id	b1	b2
+23406	b	e
+connection master;
 set global slave_exec_mode=default;
+drop table t1;
+#
+# End of 10.4 tests
+#
+# Idempotent scenario, which triggers REPLACE code to be used in the
+# event, i.e. duplicated record will be deleted and then re-inserted.
+create table t1 (i1 int, a1 text, unique key i1 (a1)) engine=myisam;
+connection slave;
+connection slave;
+set @save_slave_exec_mode= @@slave_exec_mode;
+set global slave_exec_mode = idempotent;
+insert into t1 values (1,1);
+connection master;
+insert into t1 values (2,1);
+connection slave;
+connection slave;
+select * from t1;
+i1	a1
+2	1
+connection master;
+insert into t1 values (3,3);
+update t1 set a1 = 'd' limit 1;
+update t1 set a1 = 'd2' where i1= 3;
+connection slave;
+connection slave;
+select * from t1;
+i1	a1
+2	d
+3	d2
+set global slave_exec_mode = @save_slave_exec_mode;
+connection master;
+drop table t1;
+connection slave;
+connection master;
 #
 # End of 10.4 tests
 #

mysql-test/main/long_unique_bugs_replication.test

@@ -16,17 +16,21 @@ update t1 set a1 = 'd' limit 1;
 update t1 set a1 = 'd2' where i1= 2;
 
 sync_slave_with_master;
-
+connection slave;
+select * from t1;
 connection master;
 drop table t1;
 
+sync_slave_with_master;
+connection master;
+
+
 --echo #
 --echo # MDEV-32093 long uniques break old->new replication
 --echo #
 
 # this is techically a bug in replication, but it needs an old master
 # so we'll run it as a non-replicated test with BINLOG command
-sync_slave_with_master;
 create table t1 (id int not null, b1 varchar(255) not null, b2 varchar(2550) not null, unique (id), unique key (b1,b2) using hash) default charset utf8mb3;
 set global slave_exec_mode=idempotent;
 
@@ -40,10 +44,46 @@ binlog 'bBf2ZBMBAAAANAAAAJgHAAAAAHEAAAAAAAEABHRlc3QAAnQxAAQDDw8IBP0C4h0AeqMD4A==
 ### UPDATE t1 WHERE (42127, 'b', 'e', 39952170926) SET (23406, 'b', 'e', 39952170926)
 binlog 'bBf2ZBMBAAAANAAAAHUkAAAAAHEAAAAAAAEABHRlc3QAAnQxAAQDDw8IBP0C4h0AaTGFIg==bBf2ZBgBAAAASAAAAL0kAAAAAHEAAAAAAAEABP//8I+kAAABAGIBAGWuv1VNCQAAAPBuWwAAAQBiAQBlrr9VTQkAAADxS9Lu';
 
-drop table t1;
+sync_slave_with_master;
+select * from t1;
+connection master;
 set global slave_exec_mode=default;
+drop table t1;
+
+--echo #
+--echo # End of 10.4 tests
+--echo #
+
+--echo # Idempotent scenario, which triggers REPLACE code to be used in the
+--echo # event, i.e. duplicated record will be deleted and then re-inserted.
+create table t1 (i1 int, a1 text, unique key i1 (a1)) engine=myisam;
+
+sync_slave_with_master;
+connection slave;
+set @save_slave_exec_mode= @@slave_exec_mode;
+set global slave_exec_mode = idempotent;
+insert into t1 values (1,1);
+connection master;
+insert into t1 values (2,1);
+sync_slave_with_master;
+connection slave;
+select * from t1;
+connection master;
+insert into t1 values (3,3);
+update t1 set a1 = 'd' limit 1;
+update t1 set a1 = 'd2' where i1= 3;
+sync_slave_with_master;
+
+connection slave;
+select * from t1;
+set global slave_exec_mode = @save_slave_exec_mode;
+connection master;
+drop table t1;
+sync_slave_with_master;
+connection master;
 
 --echo #
 --echo # End of 10.4 tests
 --echo #
+
 --source include/rpl_end.inc

sql/handler.cc

@@ -5297,12 +5297,6 @@ uint handler::get_dup_key(int error)
   DBUG_RETURN(errkey);
 }
 
-bool handler::has_dup_ref() const
-{
-  DBUG_ASSERT(lookup_errkey != (uint)-1 || errkey != (uint)-1);
-  return ha_table_flags() & HA_DUPLICATE_POS || lookup_errkey != (uint)-1;
-}
-
 
 /**
   Delete all files with extension from bas_ext().

sql/handler.h

@@ -3694,7 +3694,6 @@ class handler :public Sql_alloc
   virtual void print_error(int error, myf errflag);
   virtual bool get_error_message(int error, String *buf);
   uint get_dup_key(int error);
-  bool has_dup_ref() const;
   /**
     Retrieves the names of the table and the key for which there was a
     duplicate entry in the case of HA_ERR_FOREIGN_DUPLICATE_KEY.

sql/log_event.h

@@ -50,6 +50,7 @@
 #include "rpl_record.h"
 #include "rpl_reporting.h"
 #include "sql_class.h"                          /* THD */
+#include "sql_insert.h"
 #else
 typedef ulong enum_slave_exec_mode;
 #endif
@@ -4501,7 +4502,7 @@ class Table_map_log_event : public Log_event
   enum_logged_status logged_status() override { return LOGGED_TABLE_MAP; }
   bool is_valid() const override { return m_memory != NULL; /* we check malloc */ }
 
-  int get_data_size() override { return (uint) m_data_size; } 
+  int get_data_size() override { return (uint) m_data_size; }
 #ifdef MYSQL_SERVER
 #ifdef HAVE_REPLICATION
    bool is_part_of_group() override { return 1; }
@@ -4926,7 +4927,6 @@ class Rows_log_event : public Log_event
   uint find_key_parts(const KEY *key) const;
   bool use_pk_position() const;
   int find_row(rpl_group_info *);
-  int write_row(rpl_group_info *, const bool);
   int update_sequence();
 
   // Unpack the current row into m_table->record[0], but with
@@ -4990,7 +4990,8 @@ class Rows_log_event : public Log_event
       error code otherwise.
   */
   virtual
-  int do_before_row_operations(const rpl_group_info *) = 0;
+  int do_before_row_operations(const rpl_group_info *log,
+                               COPY_INFO*, Write_record*) = 0;
 
   /**
     @brief Primitive to clean up after a sequence of row executions.
@@ -5070,6 +5071,7 @@ class Write_rows_log_event : public Rows_log_event
   {
     *rows += m_row_count;
   }
+  int incomplete_record_callback(rpl_group_info *rgi);
 #endif
 
 private:
@@ -5080,7 +5082,10 @@ class Write_rows_log_event : public Rows_log_event
 #endif
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
-  int do_before_row_operations(const rpl_group_info *) override;
+  Write_record *m_write_record;
+  int write_row(rpl_group_info *, bool);
+  int do_before_row_operations(const rpl_group_info *rgi,
+                               COPY_INFO*, Write_record*) override;
   int do_after_row_operations(int) override;
   int do_exec_row(rpl_group_info *) override;
 #endif
@@ -5169,7 +5174,8 @@ class Update_rows_log_event : public Rows_log_event
 #endif
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
-  int do_before_row_operations(const rpl_group_info *) override;
+  int do_before_row_operations(const rpl_group_info *rgi,
+                               COPY_INFO*, Write_record*) override;
   int do_after_row_operations(int) override;
   int do_exec_row(rpl_group_info *) override;
 #endif /* defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) */
@@ -5260,7 +5266,8 @@ class Delete_rows_log_event : public Rows_log_event
 #endif
 
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
-  int do_before_row_operations(const rpl_group_info *const) override;
+  int do_before_row_operations(const rpl_group_info *rgi,
+                               COPY_INFO*, Write_record*) override;
   int do_after_row_operations(int) override;
   int do_exec_row(rpl_group_info *) override;
 #endif