MDEV-17614 INSERT on dup key update is replication unsafe

Problem:-
When mysql executes INSERT ON DUPLICATE KEY INSERT, the storage engine checks
if the inserted row would generate a duplicate key error. If yes, it returns
the existing row to mysql, mysql updates it and sends it back to the storage
engine.When the table has more than one unique or primary key, this statement
is sensitive to the order in which the storage engines checks the keys.
Depending on this order, the storage engine may determine different rows
to mysql, and hence mysql can update different rows.The order that the
storage engine checks keys is not deterministic. For example, InnoDB checks
keys in an order that depends on the order in which indexes were added to
the table. The first added index is checked first. So if master and slave
have added indexes in different orders, then slave may go out of sync.

Solution:-
Make INSERT...ON DUPLICATE KEY UPDATE unsafe while using stmt or mixed format
When there is more then one unique key.
Although there is two exception.
  1. Auto Increment key is not counted because Innodb will get gap lock for
    failed Insert and concurrent insert will get a next increment value. But if
    user supplies auto inc value it can be unsafe.
  2. Count only unique keys for which insertion is performed.

So this patch also addresses the bug id #72921

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

 call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
 include/master-slave.inc
 [connection master]
-call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT.");
-CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY AUTO_INCREMENT, b INT,
-UNIQUE(b));
-INSERT INTO t1(b) VALUES(1),(1),(2) ON DUPLICATE KEY UPDATE t1.b=10;
-SELECT * FROM t1;
-a	b
-1	10
-2	2
-call mtr.add_suppression("Slave SQL.*suffer.*http:..bugs.mysql.com.bug.php.id=24432");
-include/wait_for_slave_sql_error.inc [errno=1105]
-Last_SQL_Error = 'Error 'master may suffer from http://bugs.mysql.com/bug.php?id=24432 so slave stops; check error log on slave for more info' on query. Default database: 'test'. Query: 'INSERT INTO t1(b) VALUES(1),(1),(2) ON DUPLICATE KEY UPDATE t1.b=10''
-SELECT * FROM t1;
-a	b
-stop slave;
-include/wait_for_slave_to_stop.inc
-reset slave;
-reset master;
-drop table t1;
-start slave;
-include/wait_for_slave_to_start.inc
 CREATE TABLE t1 (
 id bigint(20) unsigned NOT NULL auto_increment,
 field_1 int(10) unsigned NOT NULL,

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

+include/master-slave.inc
+[connection master]
+call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY , b INT,
+UNIQUE(b),  c int) engine=innodb;
+INSERT INTO t1 VALUES (1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
+INSERT INTO t1 VALUES(2, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
+COMMIT;
+SELECT * FROM t1;
+a	b	c
+1	1	2
+2	2	3
+include/wait_for_slave_sql_error.inc [errno=1062]
+Last_SQL_Error = 'Error 'Duplicate entry '1' for key 'b'' on query. Default database: 'test'. Query: 'INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c)''
+#Different value from server
+SELECT * FROM t1;
+a	b	c
+1	1	1
+2	2	3
+stop slave;
+include/wait_for_slave_to_stop.inc
+reset slave;
+reset master;
+drop table t1;
+start slave;
+include/wait_for_slave_to_start.inc
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY auto_increment, b INT,
+UNIQUE(b),  c int) engine=innodb;
+INSERT INTO t1 VALUES (default, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (default, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+INSERT INTO t1 VALUES(default, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+COMMIT;
+SELECT * FROM t1;
+a	b	c
+1	1	2
+3	2	3
+#same data as master
+SELECT * FROM t1;
+a	b	c
+1	1	2
+3	2	3
+drop table t1;
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY, b INT,
+UNIQUE(b),  c int, d int ) engine=innodb;
+INSERT INTO t1 VALUES (1, 1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, NULL, 2, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
+INSERT INTO t1 VALUES(3, NULL, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
+COMMIT;
+SELECT * FROM t1;
+a	b	c	d
+1	1	1	1
+2	NULL	2	2
+3	NULL	2	3
+#same data as master
+SELECT * FROM t1;
+a	b	c	d
+1	1	1	1
+2	NULL	2	2
+3	NULL	2	3
+drop table t1;
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY auto_increment, b INT,
+UNIQUE(b),  c int) engine=innodb;
+INSERT INTO t1 VALUES (1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+INSERT INTO t1 VALUES(2, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+COMMIT;
+SELECT * FROM t1;
+a	b	c
+1	1	2
+2	2	3
+include/wait_for_slave_sql_error.inc [errno=1062]
+Last_SQL_Error = 'Error 'Duplicate entry '1' for key 'b'' on query. Default database: 'test'. Query: 'INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c)''
+#Different value from server
+SELECT * FROM t1;
+a	b	c
+1	1	1
+2	2	3
+stop slave;
+include/wait_for_slave_to_stop.inc
+reset slave;
+reset master;
+drop table t1;
+start slave;
+include/wait_for_slave_to_start.inc
+include/rpl_end.inc

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

 include/master-slave.inc
 [connection master]
+call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT");
 CREATE TABLE t1(id INT AUTO_INCREMENT, i INT, PRIMARY KEY (id)) ENGINE=INNODB;
 CREATE TABLE t2(id INT AUTO_INCREMENT, i INT, PRIMARY KEY (id)) ENGINE=INNODB;
 CREATE TRIGGER trig1 AFTER INSERT ON t1
@@ -43,9 +44,13 @@ include/diff_tables.inc [master:t1, slave:t1]
 DROP TABLE t1;
 CREATE TABLE t1(i INT, j INT, UNIQUE KEY(i), UNIQUE KEY(j)) ENGINE=INNODB;
 INSERT INTO t1 (i,j) VALUES (1,2) ON DUPLICATE KEY UPDATE j=j+1;
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
 START TRANSACTION;
 LOCK TABLES t1 WRITE;
 INSERT INTO t1 (i,j) VALUES (1,2) ON DUPLICATE KEY UPDATE j=j+1;
+Warnings:
+Note	1592	Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe
 UNLOCK TABLES;
 COMMIT;
 include/diff_tables.inc [master:t1, slave:t1]

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

@@ -14,45 +14,6 @@ source include/have_binlog_checksum_off.inc;
 
 source include/master-slave.inc;
 
-call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT.");
-
-#
-# This is to test that slave properly detects if
-# master may suffer from:
-# BUG#24432 "INSERT... ON DUPLICATE KEY UPDATE skips auto_increment values"
-# (i.e. on master, INSERT ON DUPLICATE KEY UPDATE is used and manipulates
-# an auto_increment column, and is binlogged statement-based).
-#
-
-# testcase with INSERT VALUES
-CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY AUTO_INCREMENT, b INT,
-UNIQUE(b));
-sync_slave_with_master;
-connection master;
-INSERT INTO t1(b) VALUES(1),(1),(2) ON DUPLICATE KEY UPDATE t1.b=10;
-SELECT * FROM t1;
-connection slave;
-
-# show the error message 
-#1105 = ER_UNKNOWN_ERROR
---let $slave_sql_errno= 1105
---let $show_slave_sql_error= 1
-call mtr.add_suppression("Slave SQL.*suffer.*http:..bugs.mysql.com.bug.php.id=24432");
---source include/wait_for_slave_sql_error.inc
-# show that it was not replicated
-SELECT * FROM t1;
-
-# restart replication for the next testcase
-stop slave;
---source include/wait_for_slave_to_stop.inc
-reset slave;
-connection master;
-reset master;
-drop table t1;
-connection slave;
-start slave;
---source include/wait_for_slave_to_start.inc
-
 # testcase with INSERT SELECT
 connection master;
 CREATE TABLE t1 (

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

+source include/have_debug.inc;
+source include/have_innodb.inc;
+-- source include/have_binlog_format_statement.inc
+source include/master-slave.inc;
+# MDEV-17614
+# INSERT on dup key update is replication unsafe
+# There can be three case
+#  1. 2 unique key, Replication is unsafe.
+#  2. 2 unique key , with one auto increment key, Safe to replicate because Innodb will acquire gap lock
+#  3. n no of unique keys (n>1) but insert is only in 1 unique key
+#  4. 2 unique key , with one auto increment key(but user gives auto inc value), unsafe to replicate
+
+# Case 1
+call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY , b INT,
+UNIQUE(b),  c int) engine=innodb;
+sync_slave_with_master;
+connection master;
+INSERT INTO t1 VALUES (1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+  --connection master1
+  INSERT INTO t1 VALUES(2, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+--connection master
+COMMIT;
+SELECT * FROM t1;
+--connection slave
+# show the error message
+--let $slave_sql_errno= 1062
+--let $show_slave_sql_error= 1
+--source include/wait_for_slave_sql_error.inc
+--echo #Different value from server
+SELECT * FROM t1;
+
+# restart replication for the next testcase
+stop slave;
+--source include/wait_for_slave_to_stop.inc
+reset slave;
+connection master;
+reset master;
+drop table t1;
+connection slave;
+start slave;
+--source include/wait_for_slave_to_start.inc
+# Case 2
+--connection master
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY auto_increment, b INT,
+UNIQUE(b),  c int) engine=innodb;
+sync_slave_with_master;
+connection master;
+INSERT INTO t1 VALUES (default, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (default, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+  --connection master1
+  INSERT INTO t1 VALUES(default, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+--connection master
+COMMIT;
+SELECT * FROM t1;
+--sync_slave_with_master
+--echo #same data as master
+SELECT * FROM t1;
+
+connection master;
+drop table t1;
+--sync_slave_with_master
+
+# Case 3
+--connection master
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY, b INT,
+UNIQUE(b),  c int, d int ) engine=innodb;
+sync_slave_with_master;
+connection master;
+INSERT INTO t1 VALUES (1, 1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, NULL, 2, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+  --connection master1
+  INSERT INTO t1 VALUES(3, NULL, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+--connection master
+COMMIT;
+SELECT * FROM t1;
+--sync_slave_with_master
+--echo #same data as master
+SELECT * FROM t1;
+connection master;
+drop table t1;
+--sync_slave_with_master
+
+# Case 4
+--connection master
+CREATE TABLE t1 (a INT NOT NULL PRIMARY KEY auto_increment, b INT,
+UNIQUE(b),  c int) engine=innodb;
+sync_slave_with_master;
+connection master;
+INSERT INTO t1 VALUES (1, 1, 1);
+BEGIN;
+INSERT INTO t1 VALUES (2, 1, 2) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+  --connection master1
+  INSERT INTO t1 VALUES(2, 2, 3) ON DUPLICATE KEY UPDATE b=VALUES(b), c=VALUES(c);
+--connection master
+COMMIT;
+SELECT * FROM t1;
+--connection slave
+# show the error message
+--let $slave_sql_errno= 1062
+--let $show_slave_sql_error= 1
+--source include/wait_for_slave_sql_error.inc
+--echo #Different value from server
+SELECT * FROM t1;
+
+# restart replication for the next testcase
+stop slave;
+--source include/wait_for_slave_to_stop.inc
+reset slave;
+connection master;
+reset master;
+drop table t1;
+connection slave;
+start slave;
+--source include/wait_for_slave_to_start.inc
+
+--source include/rpl_end.inc

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

@@ -24,7 +24,7 @@
 --source include/have_innodb.inc
 --source include/have_binlog_format_mixed.inc
 --source include/master-slave.inc
-
+call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT");
 # Case-1: BINLOG_STMT_UNSAFE_AUTOINC_COLUMNS
 #  Statement is unsafe because it invokes a trigger or a
 #  stored function that inserts into an AUTO_INCREMENT column.

sql/sql_class.cc

@@ -6144,6 +6144,48 @@ int THD::decide_logging_format(TABLE_LIST *tables)
   DBUG_RETURN(0);
 }
 
+int THD::decide_logging_format_low(TABLE *table)
+{
+  /*
+   INSERT...ON DUPLICATE KEY UPDATE on a table with more than one unique keys
+   can be unsafe.
+   */
+  if(wsrep_binlog_format() <= BINLOG_FORMAT_STMT &&
+       !is_current_stmt_binlog_format_row() &&
+       !lex->is_stmt_unsafe() &&
+       lex->sql_command == SQLCOM_INSERT &&
+       lex->duplicates == DUP_UPDATE)
+  {
+    uint unique_keys= 0;
+    uint keys= table->s->keys, i= 0;
+    Field *field;
+    for (KEY* keyinfo= table->s->key_info;
+             i < keys && unique_keys <= 1; i++, keyinfo++)
+      if (keyinfo->flags & HA_NOSAME &&
+         !(keyinfo->key_part->field->flags & AUTO_INCREMENT_FLAG &&
+             //User given auto inc can be unsafe
+             !keyinfo->key_part->field->val_int()))
+      {
+        for (uint j= 0; j < keyinfo->user_defined_key_parts; j++)
+        {
+          field= keyinfo->key_part[j].field;
+          if(!bitmap_is_set(table->write_set,field->field_index))
+            goto exit;
+        }
+        unique_keys++;
+exit:;
+      }
+
+    if (unique_keys > 1)
+    {
+      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_TWO_KEYS);
+      binlog_unsafe_warning_flags|= lex->get_stmt_unsafe_flags();
+      set_current_stmt_binlog_format_row_if_mixed();
+      return 1;
+    }
+  }
+  return 0;
+}
 
 /*
   Implementation of interface to write rows to the binary log through the

sql/sql_class.h

@@ -2208,6 +2208,20 @@ class THD :public Statement,
   /* container for handler's private per-connection data */
   Ha_data ha_data[MAX_HA];
 
+  /**
+    Bit field for the state of binlog warnings.
+
+    The first Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types of
+    unsafeness that the current statement has.
+
+    This must be a member of THD and not of LEX, because warnings are
+    detected and issued in different places (@c
+    decide_logging_format() and @c binlog_query(), respectively).
+    Between these calls, the THD->lex object may change; e.g., if a
+    stored routine is invoked.  Only THD persists between the calls.
+  */
+  uint32 binlog_unsafe_warning_flags;
+
 #ifndef MYSQL_CLIENT
   binlog_cache_mngr *  binlog_setup_trx_data();
 
@@ -2317,20 +2331,6 @@ class THD :public Statement,
   */
   enum_binlog_format current_stmt_binlog_format;
 
-  /**
-    Bit field for the state of binlog warnings.
-
-    The first Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types of
-    unsafeness that the current statement has.
-
-    This must be a member of THD and not of LEX, because warnings are
-    detected and issued in different places (@c
-    decide_logging_format() and @c binlog_query(), respectively).
-    Between these calls, the THD->lex object may change; e.g., if a
-    stored routine is invoked.  Only THD persists between the calls.
-  */
-  uint32 binlog_unsafe_warning_flags;
-
   /*
     Number of outstanding table maps, i.e., table maps in the
     transaction cache.
@@ -3939,6 +3939,18 @@ class THD :public Statement,
   }
   void leave_locked_tables_mode();
   int decide_logging_format(TABLE_LIST *tables);
+  /*
+   In Some cases when decide_logging_format is called it does not have all
+   information to decide the logging format. So that cases we call decide_logging_format_2
+   at later stages in execution.
+   One example would be binlog format for IODKU but column with unique key is not inserted.
+   We dont have inserted columns info when we call decide_logging_format so on later stage we call
+   decide_logging_format_low
+
+   @returns 0 if no format is changed
+            1 if there is change in binlog format
+  */
+  int decide_logging_format_low(TABLE *table);
 
   enum need_invoker { INVOKER_NONE=0, INVOKER_USER, INVOKER_ROLE};
   void binlog_invoker(bool role) { m_binlog_invoker= role ? INVOKER_ROLE : INVOKER_USER; }

sql/sql_insert.cc

@@ -1025,6 +1025,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
       error= 1;
       break;
     }
+    thd->decide_logging_format_low(table);
 #ifndef EMBEDDED_LIBRARY
     if (lock_type == TL_WRITE_DELAYED)
     {