Merge mysql.com:/home/marty/MySQL/mysql-4.1

into  mysql.com:/home/marty/MySQL/mysql-5.0

mysql-test/r/ndb_lock.result

@@ -63,3 +63,62 @@ pk	u	o
 5	5	5
 insert into t1 values (1,1,1);
 drop table t1;
+create table t1 (x integer not null primary key, y varchar(32)) engine = ndb;
+insert into t1 values (1,'one'), (2,'two'),(3,"three");
+begin;
+select * from t1 where x = 1 for update;
+x	y
+1	one
+begin;
+select * from t1 where x = 2 for update;
+x	y
+2	two
+select * from t1 where x = 1 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+commit;
+begin;
+select * from t1 where y = 'one' or y = 'three' for update;
+x	y
+3	three
+1	one
+begin;
+select * from t1 where x = 2 for update;
+x	y
+2	two
+select * from t1 where x = 1 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+commit;
+begin;
+select * from t1 where x = 1 lock in share mode;
+x	y
+1	one
+begin;
+select * from t1 where x = 1 lock in share mode;
+x	y
+1	one
+select * from t1 where x = 2 for update;
+x	y
+2	two
+select * from t1 where x = 1 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+commit;
+begin;
+select * from t1 where y = 'one' or y = 'three' lock in share mode;
+x	y
+3	three
+1	one
+begin;
+select * from t1 where y = 'one' lock in share mode;
+x	y
+1	one
+select * from t1 where x = 2 for update;
+x	y
+2	two
+select * from t1 where x = 1 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+commit;
+drop table t1;

mysql-test/t/ndb_lock.test

@@ -69,4 +69,80 @@ insert into t1 values (1,1,1);
 
 drop table t1;
 
+# Lock for update
+
+create table t1 (x integer not null primary key, y varchar(32)) engine = ndb;
+
+insert into t1 values (1,'one'), (2,'two'),(3,"three"); 
+
+# PK access
+connection con1;
+begin;
+select * from t1 where x = 1 for update;
+
+connection con2;
+begin;
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# scan
+connection con1;
+begin;
+select * from t1 where y = 'one' or y = 'three' for update;
+
+connection con2;
+begin;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# share locking
+
+# PK access
+connection con1;
+begin;
+select * from t1 where x = 1 lock in share mode;
+
+connection con2;
+begin;
+select * from t1 where x = 1 lock in share mode;
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# scan
+connection con1;
+begin;
+select * from t1 where y = 'one' or y = 'three' lock in share mode;
+
+connection con2;
+begin;
+select * from t1 where y = 'one' lock in share mode;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+drop table t1;
+
 # End of 4.1 tests

ndb/include/ndbapi/NdbScanOperation.hpp

@@ -61,11 +61,10 @@ public:
 #ifndef DOXYGEN_SHOULD_SKIP_DEPRECATED
   /**
    * readTuples
-   * 
    * @param lock_mode Lock mode
    * @param batch No of rows to fetch from each fragment at a time
    * @param parallel No of fragments to scan in parallell
-   * @note specifying 0 for batch and parallall means max performance
+   * @note specifying 0 for batch and parallell means max performance
    */ 
 #ifdef ndb_readtuples_impossible_overload
   int readTuples(LockMode lock_mode = LM_Read, 

ndb/src/ndbapi/NdbScanOperation.cpp

@@ -160,7 +160,7 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
     return -1;
   }
 
-  m_keyInfo = lockExcl ? 1 : 0;
+  m_keyInfo = (keyinfo || lockExcl) ? 1 : 0;
 
   bool rangeScan = false;
   if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex)
@@ -924,18 +924,28 @@ NdbScanOperation::takeOverScanOp(OperationType opType, NdbTransaction* pTrans)
     if (newOp == NULL){
       return NULL;
     }
+    if (!m_keyInfo)
+    {
+      // Cannot take over lock if no keyinfo was requested
+      setErrorCodeAbort(4604);
+      return NULL;
+    }
     pTrans->theSimpleState = 0;
     
     const Uint32 len = (tRecAttr->attrSize() * tRecAttr->arraySize() + 3)/4-1;
 
     newOp->theTupKeyLen = len;
     newOp->theOperationType = opType;
-    if (opType == DeleteRequest) {
-      newOp->theStatus = GetValue;  
-    } else {
-      newOp->theStatus = SetValue;  
+    switch (opType) {
+    case (ReadRequest):
+      newOp->theLockMode = theLockMode;
+      // Fall through
+    case (DeleteRequest):
+      newOp->theStatus = GetValue;
+      break;
+    default:
+      newOp->theStatus = SetValue;
     }
-    
     const Uint32 * src = (Uint32*)tRecAttr->aRef();
     const Uint32 tScanInfo = src[len] & 0x3FFFF;
     const Uint32 tTakeOverFragment = src[len] >> 20;

ndb/src/ndbapi/ndberror.c

@@ -290,7 +290,7 @@ ErrorBundle ErrorCodes[] = {
   { 4601, AE, "Transaction is not started"},
   { 4602, AE, "You must call getNdbOperation before executeScan" },
   { 4603, AE, "There can only be ONE operation in a scan transaction" },
-  { 4604, AE, "takeOverScanOp, opType must be UpdateRequest or DeleteRequest" },
+  { 4604, AE, "takeOverScanOp, to take over a scanned row one must explicitly request keyinfo in readTuples call" },
   { 4605, AE, "You may only call openScanRead or openScanExclusive once for each operation"},
   { 4607, AE, "There may only be one operation in a scan transaction"},
   { 4608, AE, "You can not takeOverScan unless you have used openScanExclusive"},

sql/ha_ndbcluster.h

@@ -503,6 +503,7 @@ class ha_ndbcluster: public handler
   int extra(enum ha_extra_function operation);
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
   int external_lock(THD *thd, int lock_type);
+  void unlock_row();
   int start_stmt(THD *thd, thr_lock_type lock_type);
   const char * table_type() const;
   const char ** bas_ext() const;
@@ -684,6 +685,7 @@ private:
   char m_tabname[FN_HEADLEN];
   ulong m_table_flags;
   THR_LOCK_DATA m_lock;
+  bool m_lock_tuple;
   NDB_SHARE *m_share;
   NDB_INDEX_DATA  m_index[MAX_KEY];
   // NdbRecAttr has no reference to blob