Merge bk-internal:/home/bk/mysql-5.1-new-ndb

into  ymer.(none):/usr/local/mysql/mysql-5.1-new-ndb


storage/ndb/include/ndbapi/NdbOperation.hpp:
  Auto merged
storage/ndb/include/ndbapi/NdbScanOperation.hpp:
  Auto merged
storage/ndb/src/ndbapi/NdbOperationDefine.cpp:
  Auto merged
storage/ndb/src/ndbapi/NdbScanOperation.cpp:
  Auto merged
storage/ndb/test/run-test/daily-basic-tests.txt:
  Auto merged

3 files changed, 71 insertions, 31 deletions

diff --git a/storage/ndb/src/ndbapi/NdbBlob.cpp b/storage/ndb/src/ndbapi/NdbBlob.cpp
index b3664107cce..25dcafdef53 100644
--- a/storage/ndb/src/ndbapi/NdbBlob.cpp
+++ b/storage/ndb/src/ndbapi/NdbBlob.cpp
@@ -1390,7 +1390,7 @@ NdbBlob::atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl
     if (isReadOp()) {
       // upgrade lock mode
       if (theNdbOp->theLockMode == NdbOperation::LM_CommittedRead)
-        theNdbOp->theLockMode = NdbOperation::LM_Read;
+        theNdbOp->setReadLockMode(NdbOperation::LM_Read);
       // add read of head+inline in this op
       if (getHeadInlineValue(theNdbOp) == -1)
         DBUG_RETURN(-1);
@@ -1411,7 +1411,7 @@ NdbBlob::atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl
   if (isScanOp()) {
     // upgrade lock mode
     if (theNdbOp->theLockMode == NdbOperation::LM_CommittedRead)
-      theNdbOp->theLockMode = NdbOperation::LM_Read;
+      theNdbOp->setReadLockMode(NdbOperation::LM_Read);
     // add read of head+inline in this op
     if (getHeadInlineValue(theNdbOp) == -1)
       DBUG_RETURN(-1);
diff --git a/storage/ndb/src/ndbapi/NdbOperationDefine.cpp b/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
index f818564642b..ced9b18bd55 100644
--- a/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
+++ b/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
@@ -333,6 +333,36 @@ NdbOperation::interpretedDeleteTuple()
   }//if
 }//NdbOperation::interpretedDeleteTuple()
 
+void
+NdbOperation::setReadLockMode(LockMode lockMode)
+{
+  /* We only support changing lock mode for read operations at this time. */
+  assert(theOperationType == ReadRequest || theOperationType == ReadExclusive);
+  switch (lockMode)
+  {
+    case LM_CommittedRead:
+      theOperationType= ReadRequest;
+      theSimpleIndicator= 1;
+      theDirtyIndicator= 1;
+      break;
+    case LM_Read:
+      theNdbCon->theSimpleState= 0;
+      theOperationType= ReadRequest;
+      theSimpleIndicator= 0;
+      theDirtyIndicator= 0;
+      break;
+    case LM_Exclusive:
+      theNdbCon->theSimpleState= 0;
+      theOperationType= ReadExclusive;
+      theSimpleIndicator= 0;
+      theDirtyIndicator= 0;
+      break;
+    default:
+      /* Not supported / invalid. */
+      assert(false);
+  }
+  theLockMode= lockMode;
+}
 
 
 /******************************************************************************
diff --git a/storage/ndb/src/ndbapi/NdbScanOperation.cpp b/storage/ndb/src/ndbapi/NdbScanOperation.cpp
index c253dca8c61..dc9a74ae11c 100644
--- a/storage/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/storage/ndb/src/ndbapi/NdbScanOperation.cpp
@@ -138,34 +138,9 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
   }
 
   theNdbCon->theScanningOp = this;
-  theLockMode = lm;
-
-  bool lockExcl, lockHoldMode, readCommitted;
-  switch(lm){
-  case NdbScanOperation::LM_Read:
-    lockExcl = false;
-    lockHoldMode = true;
-    readCommitted = false;
-    break;
-  case NdbScanOperation::LM_Exclusive:
-    lockExcl = true;
-    lockHoldMode = true;
-    readCommitted = false;
-    break;
-  case NdbScanOperation::LM_CommittedRead:
-    lockExcl = false;
-    lockHoldMode = false;
-    readCommitted = true;
-    break;
-  default:
-    setErrorCode(4003);
-    return -1;
-  }
-
-  m_keyInfo = ((scan_flags & SF_KeyInfo) || lockExcl) ? 1 : 0;
   bool tupScan = (scan_flags & SF_TupScan);
 
-#if 1 // XXX temp for testing
+#if 0 // XXX temp for testing
   { char* p = getenv("NDB_USE_TUPSCAN");
     if (p != 0) {
       unsigned n = atoi(p); // 0-10
@@ -225,13 +200,13 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
   Uint32 reqInfo = 0;
   ScanTabReq::setParallelism(reqInfo, parallel);
   ScanTabReq::setScanBatch(reqInfo, 0);
-  ScanTabReq::setLockMode(reqInfo, lockExcl);
-  ScanTabReq::setHoldLockFlag(reqInfo, lockHoldMode);
-  ScanTabReq::setReadCommittedFlag(reqInfo, readCommitted);
   ScanTabReq::setRangeScanFlag(reqInfo, rangeScan);
   ScanTabReq::setTupScanFlag(reqInfo, tupScan);
   req->requestInfo = reqInfo;
 
+  m_keyInfo = (scan_flags & SF_KeyInfo) ? 1 : 0;
+  setReadLockMode(lm);
+
   Uint64 transId = theNdbCon->getTransactionId();
   req->transId1 = (Uint32) transId;
   req->transId2 = (Uint32) (transId >> 32);
@@ -251,6 +226,41 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
   return 0;
 }
 
+void
+NdbScanOperation::setReadLockMode(LockMode lockMode)
+{
+  bool lockExcl, lockHoldMode, readCommitted;
+  switch (lockMode)
+  {
+    case LM_CommittedRead:
+      lockExcl= false;
+      lockHoldMode= false;
+      readCommitted= true;
+      break;
+    case LM_Read:
+      lockExcl= false;
+      lockHoldMode= true;
+      readCommitted= false;
+      break;
+    case LM_Exclusive:
+      lockExcl= true;
+      lockHoldMode= true;
+      readCommitted= false;
+      m_keyInfo= 1;
+      break;
+    default:
+      /* Not supported / invalid. */
+      assert(false);
+  }
+  theLockMode= lockMode;
+  ScanTabReq *req= CAST_PTR(ScanTabReq, theSCAN_TABREQ->getDataPtrSend());
+  Uint32 reqInfo= req->requestInfo;
+  ScanTabReq::setLockMode(reqInfo, lockExcl);
+  ScanTabReq::setHoldLockFlag(reqInfo, lockHoldMode);
+  ScanTabReq::setReadCommittedFlag(reqInfo, readCommitted);
+  req->requestInfo= reqInfo;
+}
+
 int
 NdbScanOperation::fix_receivers(Uint32 parallel){
   assert(parallel > 0);