Fix for BUG#25507 "multi-row insert delayed + auto increment causes

duplicate key entries on slave" (two concurrrent connections doing
multi-row INSERT DELAYED to insert into an auto_increment column,
caused replication slave to stop with "duplicate key error" (and
binlog was wrong)), and BUG#26116 "If multi-row INSERT
DELAYED has errors, statement-based binlogging breaks" (the binlog
was not accounting for all rows inserted, or slave could stop).
The fix is that: if (statement-based) binlogging is on, a multi-row
INSERT DELAYED is silently converted to a non-delayed INSERT.
Note: it is not possible to test BUG#25507 in 5.0 (requires mysqlslap),
so it is tested only in the changeset for 5.1. However, BUG#26116
is tested here, and the fix for BUG#25507 is the same code change.

mysql-test/r/innodb-replace.result

@@ -2,11 +2,11 @@ drop table if exists t1;
 create table t1 (c1 char(5) unique not null, c2 int, stamp timestamp) engine=innodb;
 select * from t1;
 c1	c2	stamp
-replace delayed into t1 (c1, c2)  values ( "text1","11"),( "text2","12");
+replace delayed into t1 (c1, c2)  values ( "text1","11");
 ERROR HY000: Table storage engine for 't1' doesn't have this option
 select * from t1;
 c1	c2	stamp
-replace delayed into t1 (c1, c2)  values ( "text1","12"),( "text2","13"),( "text3","14", "a" ),( "text4","15", "b" );
+replace delayed into t1 (c1, c2)  values ( "text1","12");
 ERROR HY000: Table storage engine for 't1' doesn't have this option
 select * from t1;
 c1	c2	stamp

mysql-test/r/rpl_insert_delayed.result

@@ -0,0 +1,31 @@
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+CREATE TABLE t1 (id INT primary key auto_increment, name VARCHAR(64));
+truncate table t1;
+insert delayed into t1 values(10, "my name");
+insert delayed into t1 values(10, "is Bond"), (20, "James Bond");
+ERROR 23000: Duplicate entry '10' for key 1
+flush table t1;
+select * from t1;
+id	name
+10	my name
+select * from t1;
+id	name
+10	my name
+delete from t1 where id!=10;
+insert delayed into t1 values(20, "is Bond"), (10, "James Bond");
+ERROR 23000: Duplicate entry '10' for key 1
+flush table t1;
+select * from t1;
+id	name
+10	my name
+20	is Bond
+select * from t1;
+id	name
+10	my name
+20	is Bond
+drop table t1;

mysql-test/t/innodb-replace.test

@@ -12,10 +12,10 @@ drop table if exists t1;
 create table t1 (c1 char(5) unique not null, c2 int, stamp timestamp) engine=innodb;
 select * from t1;
 --error 1031
-replace delayed into t1 (c1, c2)  values ( "text1","11"),( "text2","12");
+replace delayed into t1 (c1, c2)  values ( "text1","11");
 select * from t1;
 --error 1031
-replace delayed into t1 (c1, c2)  values ( "text1","12"),( "text2","13"),( "text3","14", "a" ),( "text4","15", "b" );
+replace delayed into t1 (c1, c2)  values ( "text1","12");
 select * from t1;
 drop table t1;
 

mysql-test/t/rpl_insert_delayed.test

@@ -0,0 +1,67 @@
+--source include/master-slave.inc
+--source include/not_embedded.inc
+--source include/not_windows.inc
+
+connection master;
+
+let $binlog_format_statement=1;
+
+CREATE TABLE t1 (id INT primary key auto_increment, name VARCHAR(64));
+
+sync_slave_with_master;
+
+#
+# BUG#26116 "If multi-row INSERT DELAYED has errors,
+# statement-based binlogging breaks";
+# happened only in statement-based binlogging.
+#
+
+connection master;
+truncate table t1;
+# first scenario: duplicate on first row
+insert delayed into t1 values(10, "my name");
+if ($binlog_format_statement)
+{
+  # statement below will be converted to non-delayed INSERT and so
+  # will stop at first error, guaranteeing replication.
+  --error ER_DUP_ENTRY
+  insert delayed into t1 values(10, "is Bond"), (20, "James Bond");
+}
+if (!$binlog_format_statement)
+{
+  insert delayed into t1 values(10, "is Bond"), (20, "James Bond");
+}
+flush table t1; # to wait for INSERT DELAYED to be done
+select * from t1;
+sync_slave_with_master;
+# when bug existed in statement-based binlogging, t1 on slave had
+# different content from on master
+select * from t1;
+
+# second scenario: duplicate on second row
+connection master;
+delete from t1 where id!=10;
+if ($binlog_format_statement)
+{
+  # statement below will be converted to non-delayed INSERT and so
+  # will be binlogged with its ER_DUP_ENTRY error code, guaranteeing
+  # replication (slave will hit the same error code and so be fine).
+  --error ER_DUP_ENTRY
+  insert delayed into t1 values(20, "is Bond"), (10, "James Bond");
+}
+if (!$binlog_format_statement)
+{
+  insert delayed into t1 values(20, "is Bond"), (10, "James Bond");
+}
+flush table t1; # to wait for INSERT DELAYED to be done
+select * from t1;
+sync_slave_with_master;
+# when bug existed in statement-based binlogging, query was binlogged
+# with error_code=0 so slave stopped
+select * from t1;
+
+# clean up
+connection master;
+drop table t1;
+sync_slave_with_master;
+connection master;

sql/sql_insert.cc

@@ -356,6 +356,27 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
       (duplic == DUP_UPDATE))
     lock_type=TL_WRITE;
 #endif
+  if ((lock_type == TL_WRITE_DELAYED) &&
+      log_on && mysql_bin_log.is_open() &&
+      (values_list.elements > 1))
+  {
+    /*
+      Statement-based binary logging does not work in this case, because:
+      a) two concurrent statements may have their rows intermixed in the
+      queue, leading to autoincrement replication problems on slave (because
+      the values generated used for one statement don't depend only on the
+      value generated for the first row of this statement, so are not
+      replicable)
+      b) if first row of the statement has an error the full statement is
+      not binlogged, while next rows of the statement may be inserted.
+      c) if first row succeeds, statement is binlogged immediately with a
+      zero error code (i.e. "no error"), if then second row fails, query
+      will fail on slave too and slave will stop (wrongly believing that the
+      master got no error).
+      So we fallback to non-delayed INSERT.
+    */
+    lock_type= TL_WRITE;
+  }
   table_list->lock_type= lock_type;
 
 #ifndef EMBEDDED_LIBRARY