Bug #28719: multi pk update ignore corrupts data

- check multi update as well as update
- this bug is not present in 5.0, but execution patch is wrong, so there are probably other bugs

mysql-test/r/ndb_basic.result

@@ -770,4 +770,75 @@ c	abc	ab
 d	ab	ab
 e	abc	abc
 DROP TABLE t1;
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+create table t2 (a int not null primary key, b int not null) engine=ndb;
+insert into t1 values (1,10), (2,20), (3,30);
+insert into t2 values (1,10), (2,20), (3,30);
+select * from t1 order by a;
+a	b
+1	10
+2	20
+3	30
+delete from t1 where a > 0 order by a desc limit 1;
+select * from t1 order by a;
+a	b
+1	10
+2	20
+delete from t1,t2 using t1,t2 where t1.a = t2.a;
+select * from t2 order by a;
+a	b
+3	30
+drop table t1,t2;
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+insert into t1 values (1,10), (2,20), (3,30);
+insert into t1 set a=1, b=100;
+ERROR 23000: Duplicate entry '1' for key 1
+insert ignore into t1 set a=1, b=100;
+select * from t1 order by a;
+a	b
+1	10
+2	20
+3	30
+insert into t1 set a=1, b=1000 on duplicate key update b=b+1;
+select * from t1 order by a;
+a	b
+1	11
+2	20
+3	30
+drop table t1;
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+create table t2 (c int not null primary key, d int not null) engine=ndb;
+insert into t1 values (1,10), (2,10), (3,30), (4, 30);
+insert into t2 values (1,10), (2,10), (3,30), (4, 30);
+update t1 set a = 1 where a = 3;
+ERROR 23000: Duplicate entry '1' for key 1
+select * from t1 order by a;
+a	b
+1	10
+2	10
+3	30
+4	30
+update t1 set b = 1 where a > 1 order by a desc limit 1;
+select * from t1 order by a;
+a	b
+1	10
+2	10
+3	30
+4	1
+update t1,t2 set a = 1, c = 1 where a = 3 and c = 3;
+ERROR 23000: Duplicate entry '1' for key 1
+select * from t1 order by a;
+a	b
+1	10
+2	10
+3	30
+4	1
+update ignore t1,t2 set a = 1, c = 1 where a = 3 and c = 3;
+select * from t1 order by a;
+a	b
+1	10
+2	10
+3	30
+4	1
+drop table t1,t2;
 End of 5.0 tests

mysql-test/t/ndb_basic.test

@@ -740,6 +740,46 @@ INSERT INTO t1 VALUES
 SELECT * FROM t1 ORDER BY a;
 DROP TABLE t1;
 
+# delete
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+create table t2 (a int not null primary key, b int not null) engine=ndb;
+insert into t1 values (1,10), (2,20), (3,30);
+insert into t2 values (1,10), (2,20), (3,30);
+select * from t1 order by a;
+delete from t1 where a > 0 order by a desc limit 1;
+select * from t1 order by a;
+delete from t1,t2 using t1,t2 where t1.a = t2.a;
+select * from t2 order by a;
+drop table t1,t2;
+
+# insert ignore
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+insert into t1 values (1,10), (2,20), (3,30);
+--error ER_DUP_ENTRY
+insert into t1 set a=1, b=100;
+insert ignore into t1 set a=1, b=100;
+select * from t1 order by a;
+insert into t1 set a=1, b=1000 on duplicate key update b=b+1;
+select * from t1 order by a;
+drop table t1;
+
+# update
+create table t1 (a int not null primary key, b int not null) engine=ndb;
+create table t2 (c int not null primary key, d int not null) engine=ndb;
+insert into t1 values (1,10), (2,10), (3,30), (4, 30);
+insert into t2 values (1,10), (2,10), (3,30), (4, 30);
+--error ER_DUP_ENTRY
+update t1 set a = 1 where a = 3;
+select * from t1 order by a;
+update t1 set b = 1 where a > 1 order by a desc limit 1;
+select * from t1 order by a;
+--error ER_DUP_ENTRY
+update t1,t2 set a = 1, c = 1 where a = 3 and c = 3;
+select * from t1 order by a;
+update ignore t1,t2 set a = 1, c = 1 where a = 3 and c = 3;
+select * from t1 order by a;
+drop table t1,t2;
+
 # End of 5.0 tests
 --echo End of 5.0 tests
 

sql/ha_ndbcluster.cc

@@ -2449,7 +2449,8 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
    * If IGNORE the ignore constraint violations on primary and unique keys,
    * but check that it is not part of INSERT ... ON DUPLICATE KEY UPDATE
    */
-  if (m_ignore_dup_key && thd->lex->sql_command == SQLCOM_UPDATE)
+  if (m_ignore_dup_key && (thd->lex->sql_command == SQLCOM_UPDATE ||
+                           thd->lex->sql_command == SQLCOM_UPDATE_MULTI))
   {
     int peek_res= peek_indexed_rows(new_data, pk_update);