Merge mskold@bk-internal.mysql.com:/home/bk/mysql-4.1

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

6 files changed, 62 insertions, 15 deletions

diff --git a/ndb/include/ndbapi/NdbIndexScanOperation.hpp b/ndb/include/ndbapi/NdbIndexScanOperation.hpp
index 7cd2daea6a6..e96f46e0f32 100644
--- a/ndb/include/ndbapi/NdbIndexScanOperation.hpp
+++ b/ndb/include/ndbapi/NdbIndexScanOperation.hpp
@@ -45,14 +45,15 @@ public:
   NdbResultSet* readTuples(LockMode = LM_Read,
 			   Uint32 batch = 0, 
 			   Uint32 parallel = 0,
-			   bool order_by = false);
+			   bool order_by = false,
+			   bool keyinfo = false);
   
   inline NdbResultSet* readTuples(int parallell){
-    return readTuples(LM_Read, 0, parallell, false);
+    return readTuples(LM_Read, 0, parallell);
   }
   
   inline NdbResultSet* readTuplesExclusive(int parallell = 0){
-    return readTuples(LM_Exclusive, 0, parallell, false);
+    return readTuples(LM_Exclusive, 0, parallell);
   }
 
   /**
diff --git a/ndb/include/ndbapi/NdbResultSet.hpp b/ndb/include/ndbapi/NdbResultSet.hpp
index dc0288a380c..2a196b1c6ae 100644
--- a/ndb/include/ndbapi/NdbResultSet.hpp
+++ b/ndb/include/ndbapi/NdbResultSet.hpp
@@ -102,6 +102,27 @@ public:
   int restart(bool forceSend = false);
   
   /**
+   * Lock current row by transfering scan operation to a locking transaction. 
+   * Use this function 
+   * when a scan has found a record that you want to lock. 
+   * 1. Start a new transaction.
+   * 2. Call the function takeOverForUpdate using your new transaction 
+   *    as parameter, all the properties of the found record will be copied 
+   *    to the new transaction.
+   * 3. When you execute the new transaction, the lock held by the scan will 
+   *    be transferred to the new transaction(it's taken over).
+   *
+   * @note You must have started the scan with openScanExclusive
+   *       or explictly have requested keyinfo to be able to lock 
+   *       the found tuple.
+   *
+   * @param lockingTrans the locking transaction connection.
+   * @return an NdbOperation or NULL.
+   */
+  NdbOperation* lockTuple();
+  NdbOperation*	lockTuple(NdbConnection* lockingTrans);
+
+  /**
    * Transfer scan operation to an updating transaction. Use this function 
    * when a scan has found a record that you want to update. 
    * 1. Start a new transaction.
diff --git a/ndb/include/ndbapi/NdbScanOperation.hpp b/ndb/include/ndbapi/NdbScanOperation.hpp
index f6e68dd4abe..af656720efb 100644
--- a/ndb/include/ndbapi/NdbScanOperation.hpp
+++ b/ndb/include/ndbapi/NdbScanOperation.hpp
@@ -64,14 +64,16 @@ public:
    * Tuples are not stored in NdbResultSet until execute(NoCommit) 
    * has been executed and nextResult has been called.
    * 
+   * @param keyinfo   Return primary key, needed to be able to call lockTuple
    * @param parallel  Scan parallelism
    * @param batch No of rows to fetch from each fragment at a time
    * @param LockMode  Scan lock handling   
    * @returns NdbResultSet.
-   * @note specifying 0 for batch and parallall means max performance
+   * @note specifying 0 for batch and parallell means max performance
    */ 
   NdbResultSet* readTuples(LockMode = LM_Read, 
-			   Uint32 batch = 0, Uint32 parallel = 0);
+			   Uint32 batch = 0, Uint32 parallel = 0, 
+			   bool keyinfo = false);
   
   inline NdbResultSet* readTuples(int parallell){
     return readTuples(LM_Read, 0, parallell);
diff --git a/ndb/src/ndbapi/NdbResultSet.cpp b/ndb/src/ndbapi/NdbResultSet.cpp
index 87b304126ba..780660169b3 100644
--- a/ndb/src/ndbapi/NdbResultSet.cpp
+++ b/ndb/src/ndbapi/NdbResultSet.cpp
@@ -73,6 +73,17 @@ void NdbResultSet::close(bool forceSend)
 }
 
 NdbOperation* 
+NdbResultSet::lockTuple(){
+  return lockTuple(m_operation->m_transConnection);
+}
+
+NdbOperation* 
+NdbResultSet::lockTuple(NdbConnection* takeOverTrans){
+  return m_operation->takeOverScanOp(NdbOperation::ReadRequest, 
+				     takeOverTrans);
+}
+
+NdbOperation* 
 NdbResultSet::updateTuple(){
   return updateTuple(m_operation->m_transConnection);
 }
diff --git a/ndb/src/ndbapi/NdbScanOperation.cpp b/ndb/src/ndbapi/NdbScanOperation.cpp
index fc5a22cce17..0a39651ce28 100644
--- a/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/ndb/src/ndbapi/NdbScanOperation.cpp
@@ -126,7 +126,8 @@ NdbScanOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection)
 
 NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
 					   Uint32 batch, 
-					   Uint32 parallel)
+					   Uint32 parallel,
+					   bool keyinfo)
 {
   m_ordered = 0;
 
@@ -170,7 +171,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
     return 0;
   }
 
-  m_keyInfo = lockExcl ? 1 : 0;
+  m_keyInfo = (keyinfo || lockExcl) ? 1 : 0;
 
   bool range = false;
   if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex ||
@@ -956,18 +957,28 @@ NdbScanOperation::takeOverScanOp(OperationType opType, NdbConnection* 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 tTakeOverNode = src[len] >> 20;
@@ -1241,8 +1252,9 @@ NdbResultSet*
 NdbIndexScanOperation::readTuples(LockMode lm,
 				  Uint32 batch,
 				  Uint32 parallel,
-				  bool order_by){
-  NdbResultSet * rs = NdbScanOperation::readTuples(lm, batch, 0);
+				  bool order_by,
+				  bool keyinfo){
+  NdbResultSet * rs = NdbScanOperation::readTuples(lm, batch, 0, keyinfo);
   if(rs && order_by){
     m_ordered = 1;
     Uint32 cnt = m_accessTable->getNoOfColumns() - 1;
diff --git a/ndb/src/ndbapi/ndberror.c b/ndb/src/ndbapi/ndberror.c
index 69fc47ff70c..dea1b8c40ae 100644
--- a/ndb/src/ndbapi/ndberror.c
+++ b/ndb/src/ndbapi/ndberror.c
@@ -281,7 +281,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"},