Bug#6020, any lock >= write_allow_write is a write lock

mysql-test/r/ndb_lock.result

@@ -28,3 +28,38 @@ x	y
 2	two
 3	three
 commit;
+drop table t1;
+create table t1 (pk integer not null primary key, u int not null, o int not null, 
+unique(u), key(o)) engine = ndb;
+insert into t1 values (1,1,1), (2,2,2), (3,3,3), (4,4,4), (5,5,5);
+lock tables t1 write;
+delete from t1 where pk = 1;
+unlock tables;
+select * from t1 order by pk;
+pk	u	o
+2	2	2
+3	3	3
+4	4	4
+5	5	5
+insert into t1 values (1,1,1);
+lock tables t1 write;
+delete from t1 where u = 1;
+unlock tables;
+select * from t1 order by pk;
+pk	u	o
+2	2	2
+3	3	3
+4	4	4
+5	5	5
+insert into t1 values (1,1,1);
+lock tables t1 write;
+delete from t1 where o = 1;
+unlock tables;
+select * from t1 order by pk;
+pk	u	o
+2	2	2
+3	3	3
+4	4	4
+5	5	5
+insert into t1 values (1,1,1);
+drop table t1;

mysql-test/t/ndb_lock.test

@@ -39,3 +39,32 @@ commit;
 connection con2;
 select * from t1 order by x;
 commit;
+
+drop table t1;
+
+###
+# Bug#6020
+create table t1 (pk integer not null primary key, u int not null, o int not null, 
+                 unique(u), key(o)) engine = ndb;
+insert into t1 values (1,1,1), (2,2,2), (3,3,3), (4,4,4), (5,5,5);
+
+lock tables t1 write;
+delete from t1 where pk = 1;
+unlock tables;
+select * from t1 order by pk;
+insert into t1 values (1,1,1);
+
+lock tables t1 write;
+delete from t1 where u = 1;
+unlock tables;
+select * from t1 order by pk;
+insert into t1 values (1,1,1);
+
+lock tables t1 write;
+delete from t1 where o = 1;
+unlock tables;
+select * from t1 order by pk;
+insert into t1 values (1,1,1);
+
+drop table t1;
+

sql/ha_ndbcluster.cc

@@ -824,7 +824,7 @@ void ha_ndbcluster::release_metadata()
 
 int ha_ndbcluster::get_ndb_lock_type(enum thr_lock_type type)
 {
-  if (type == TL_WRITE_ALLOW_WRITE)
+  if (type >= TL_WRITE_ALLOW_WRITE)
     return NdbOperation::LM_Exclusive;
   else if (uses_blob_value(retrieve_all_fields))
     /*
@@ -1161,7 +1161,7 @@ inline int ha_ndbcluster::next_result(byte *buf)
      If this an update or delete, call nextResult with false
      to process any records already cached in NdbApi
   */
-  bool contact_ndb= m_lock.type != TL_WRITE_ALLOW_WRITE;
+  bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE;
   do {
     DBUG_PRINT("info", ("Call nextResult, contact_ndb: %d", contact_ndb));
     /*
@@ -2731,6 +2731,9 @@ THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
     /* If we are not doing a LOCK TABLE, then allow multiple
        writers */
     
+    /* Since NDB does not currently have table locks
+       this is treated as a ordinary lock */
+
     if ((lock_type >= TL_WRITE_ALLOW_WRITE &&
          lock_type <= TL_WRITE) && !thd->in_lock_tables)      
       lock_type= TL_WRITE_ALLOW_WRITE;