Merge mysql.com:/home/stewart/Documents/MySQL/5.0/merge-queue

into  mysql.com:/home/stewart/Documents/MySQL/5.0/main

10 files changed, 131 insertions, 82 deletions

diff --git a/ndb/include/ndbapi/NdbIndexScanOperation.hpp b/ndb/include/ndbapi/NdbIndexScanOperation.hpp
index e9f92d84d1c..638ecb17779 100644
--- a/ndb/include/ndbapi/NdbIndexScanOperation.hpp
+++ b/ndb/include/ndbapi/NdbIndexScanOperation.hpp
@@ -61,11 +61,14 @@ public:
                         Uint32 parallel,
                         bool order_by,
                         bool order_desc = false,
-                        bool read_range_no = false) {
+                        bool read_range_no = false,
+			bool keyinfo = false) {
     Uint32 scan_flags =
       (SF_OrderBy & -(Int32)order_by) |
       (SF_Descending & -(Int32)order_desc) |
-      (SF_ReadRangeNo & -(Int32)read_range_no);
+      (SF_ReadRangeNo & -(Int32)read_range_no) | 
+      (SF_KeyInfo & -(Int32)keyinfo);
+    
     return readTuples(lock_mode, scan_flags, parallel);
   }
 #endif
diff --git a/ndb/include/ndbapi/NdbScanOperation.hpp b/ndb/include/ndbapi/NdbScanOperation.hpp
index 77b255dc2f4..4a8425852b9 100644
--- a/ndb/include/ndbapi/NdbScanOperation.hpp
+++ b/ndb/include/ndbapi/NdbScanOperation.hpp
@@ -44,7 +44,8 @@ public:
     SF_TupScan = (1 << 16),     // scan TUP - only LM_CommittedRead
     SF_OrderBy = (1 << 24),     // index scan in order
     SF_Descending = (2 << 24),  // index scan in descending order
-    SF_ReadRangeNo = (4 << 24)  // enable @ref get_range_no
+    SF_ReadRangeNo = (4 << 24), // enable @ref get_range_no
+    SF_KeyInfo = 1              // request KeyInfo to be sent back
   };
 
   /**
@@ -61,15 +62,14 @@ 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, 
-		 Uint32 batch = 0, Uint32 parallel = 0);
+		 Uint32 batch = 0, Uint32 parallel = 0, bool keyinfo = false);
 #endif
   
   inline int readTuples(int parallell){
@@ -142,6 +142,20 @@ public:
   void close(bool forceSend = false, bool releaseOp = false);
 
   /**
+   * Lock current tuple
+   *
+   * @return an NdbOperation or NULL.
+   */
+  NdbOperation* lockCurrentTuple();
+  /**
+   * Lock current tuple
+   *
+   * @param lockTrans Transaction that should perform the lock
+   *
+   * @return an NdbOperation or NULL.
+   */
+  NdbOperation*	lockCurrentTuple(NdbTransaction* lockTrans);
+  /**
    * Update current tuple
    *
    * @return an NdbOperation or NULL.
@@ -251,6 +265,19 @@ protected:
 
 inline
 NdbOperation* 
+NdbScanOperation::lockCurrentTuple(){
+  return lockCurrentTuple(m_transConnection);
+}
+
+inline
+NdbOperation* 
+NdbScanOperation::lockCurrentTuple(NdbTransaction* takeOverTrans){
+  return takeOverScanOp(NdbOperation::ReadRequest, 
+			takeOverTrans);
+}
+
+inline
+NdbOperation* 
 NdbScanOperation::updateCurrentTuple(){
   return updateCurrentTuple(m_transConnection);
 }
diff --git a/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp b/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
index 908e2a81345..d2f9150ade0 100644
--- a/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
+++ b/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
@@ -234,6 +234,7 @@ void Cmvmi::execEVENT_REP(Signal* signal)
 void
 Cmvmi::execEVENT_SUBSCRIBE_REQ(Signal * signal){
   EventSubscribeReq * subReq = (EventSubscribeReq *)&signal->theData[0];
+  Uint32 senderRef = signal->getSendersBlockRef();
   SubscriberPtr ptr;
   jamEntry();
   DBUG_ENTER("Cmvmi::execEVENT_SUBSCRIBE_REQ");
@@ -251,7 +252,7 @@ Cmvmi::execEVENT_SUBSCRIBE_REQ(Signal * signal){
      * Create a new one
      */
     if(subscribers.seize(ptr) == false){
-      sendSignal(subReq->blockRef, GSN_EVENT_SUBSCRIBE_REF, signal, 1, JBB);
+      sendSignal(senderRef, GSN_EVENT_SUBSCRIBE_REF, signal, 1, JBB);
       return;
     }
     ptr.p->logLevel.clear();
@@ -278,7 +279,7 @@ Cmvmi::execEVENT_SUBSCRIBE_REQ(Signal * signal){
   }
   
   signal->theData[0] = ptr.i;
-  sendSignal(ptr.p->blockRef, GSN_EVENT_SUBSCRIBE_CONF, signal, 1, JBB);
+  sendSignal(senderRef, GSN_EVENT_SUBSCRIBE_CONF, signal, 1, JBB);
   DBUG_VOID_RETURN;
 }
 
diff --git a/ndb/src/mgmsrv/MgmtSrvr.cpp b/ndb/src/mgmsrv/MgmtSrvr.cpp
index 41359fe65bd..2505515c086 100644
--- a/ndb/src/mgmsrv/MgmtSrvr.cpp
+++ b/ndb/src/mgmsrv/MgmtSrvr.cpp
@@ -119,41 +119,50 @@ MgmtSrvr::logLevelThreadRun()
     /**
      * Handle started nodes
      */
-    EventSubscribeReq req;
-    req = m_event_listner[0].m_logLevel;
-    req.blockRef = _ownReference;
-
-    SetLogLevelOrd ord;
-    
     m_started_nodes.lock();
-    while(m_started_nodes.size() > 0){
-      Uint32 node = m_started_nodes[0];
-      m_started_nodes.erase(0, false);
-      m_started_nodes.unlock();
+    if (m_started_nodes.size() > 0)
+    {
+      // calculate max log level
+      EventSubscribeReq req;
+      {
+        LogLevel tmp;
+        m_event_listner.lock();
+        for(int i = m_event_listner.m_clients.size() - 1; i >= 0; i--)
+          tmp.set_max(m_event_listner[i].m_logLevel);
+        m_event_listner.unlock();
+        req = tmp;
+      }
+      req.blockRef = _ownReference;
+      while (m_started_nodes.size() > 0)
+      {
+        Uint32 node = m_started_nodes[0];
+        m_started_nodes.erase(0, false);
+        m_started_nodes.unlock();
 
-      setEventReportingLevelImpl(node, req);
-      
-      ord = m_nodeLogLevel[node];
-      setNodeLogLevelImpl(node, ord);
-      
-      m_started_nodes.lock();
-    }				 
+        setEventReportingLevelImpl(node, req);
+
+        SetLogLevelOrd ord;
+        ord = m_nodeLogLevel[node];
+        setNodeLogLevelImpl(node, ord);
+
+        m_started_nodes.lock();
+      }
+    }
     m_started_nodes.unlock();
     
     m_log_level_requests.lock();
-    while(m_log_level_requests.size() > 0){
-      req = m_log_level_requests[0];
+    while (m_log_level_requests.size() > 0)
+    {
+      EventSubscribeReq req = m_log_level_requests[0];
       m_log_level_requests.erase(0, false);
       m_log_level_requests.unlock();
-      
-      LogLevel tmp;
-      tmp = req;
-      
+
       if(req.blockRef == 0){
 	req.blockRef = _ownReference;
 	setEventReportingLevelImpl(0, req);
       } else {
-	ord = req;
+        SetLogLevelOrd ord;
+        ord = req;
 	setNodeLogLevelImpl(req.blockRef, ord);
       }
       m_log_level_requests.lock();
@@ -1538,7 +1547,8 @@ int
 MgmtSrvr::setEventReportingLevelImpl(int nodeId, 
 				     const EventSubscribeReq& ll)
 {
-  INIT_SIGNAL_SENDER(ss,nodeId);
+  SignalSender ss(theFacade);
+  ss.lock();
 
   SimpleSignal ssig;
   EventSubscribeReq * dst = 
@@ -1547,41 +1557,54 @@ MgmtSrvr::setEventReportingLevelImpl(int nodeId,
 	   EventSubscribeReq::SignalLength);
   *dst = ll;
 
-  send(ss,ssig,nodeId,NODE_TYPE_DB);
+  NodeBitmask nodes;
+  nodes.clear();
+  Uint32 max = (nodeId == 0) ? (nodeId = 1, MAX_NDB_NODES) : nodeId;
+  for(; nodeId <= max; nodeId++)
+  {
+    if (nodeTypes[nodeId] != NODE_TYPE_DB)
+      continue;
+    if (okToSendTo(nodeId, true))
+      continue;
+    if (ss.sendSignal(nodeId, &ssig) == SEND_OK)
+    {
+      nodes.set(nodeId);
+    }
+  }
 
-#if 0
-  while (1)
+  int error = 0;
+  while (!nodes.isclear())
   {
     SimpleSignal *signal = ss.waitFor();
     int gsn = signal->readSignalNumber();
-    switch (gsn) { 
+    nodeId = refToNode(signal->header.theSendersBlockRef);
+    switch (gsn) {
     case GSN_EVENT_SUBSCRIBE_CONF:{
+      nodes.clear(nodeId);
       break;
     }
     case GSN_EVENT_SUBSCRIBE_REF:{
-      return SEND_OR_RECEIVE_FAILED;
+      nodes.clear(nodeId);
+      error = 1;
+      break;
     }
     case GSN_NF_COMPLETEREP:{
       const NFCompleteRep * const rep =
 	CAST_CONSTPTR(NFCompleteRep, signal->getDataPtr());
-      if (rep->failedNodeId == nodeId)
-	return SEND_OR_RECEIVE_FAILED;
+      nodes.clear(rep->failedNodeId);
       break;
     }
     case GSN_NODE_FAILREP:{
-      const NodeFailRep * const rep =
-	CAST_CONSTPTR(NodeFailRep, signal->getDataPtr());
-      if (NodeBitmask::get(rep->theNodes,nodeId))
-	return SEND_OR_RECEIVE_FAILED;
+      // ignore, NF_COMPLETEREP will arrive later
       break;
     }
     default:
       report_unknown_signal(signal);
       return SEND_OR_RECEIVE_FAILED;
     }
-
   }
-#endif
+  if (error)
+    return SEND_OR_RECEIVE_FAILED;
   return 0;
 }
 
@@ -1601,19 +1624,6 @@ MgmtSrvr::setNodeLogLevelImpl(int nodeId, const SetLogLevelOrd & ll)
   return ss.sendSignal(nodeId, &ssig) == SEND_OK ? 0 : SEND_OR_RECEIVE_FAILED;
 }
 
-int
-MgmtSrvr::send(SignalSender &ss, SimpleSignal &ssig, Uint32 node, Uint32 node_type){
-  Uint32 max = (node == 0) ? MAX_NODES : node + 1;
-  
-  for(; node < max; node++){
-    while(nodeTypes[node] != (int)node_type && node < max) node++;
-    if(nodeTypes[node] != (int)node_type)
-      break;
-    ss.sendSignal(node, &ssig);
-  }
-  return 0;
-}
-
 //****************************************************************************
 //****************************************************************************
 
diff --git a/ndb/src/mgmsrv/MgmtSrvr.hpp b/ndb/src/mgmsrv/MgmtSrvr.hpp
index 2b84eac6d2c..7811cf0e5d1 100644
--- a/ndb/src/mgmsrv/MgmtSrvr.hpp
+++ b/ndb/src/mgmsrv/MgmtSrvr.hpp
@@ -494,8 +494,6 @@ public:
 private:
   //**************************************************************************
 
-  int send(SignalSender &ss, SimpleSignal &ssig, Uint32 node, Uint32 node_type);
-
   int sendStopMgmd(NodeId nodeId,
                    bool abort,
                    bool stop,
diff --git a/ndb/src/mgmsrv/Services.cpp b/ndb/src/mgmsrv/Services.cpp
index 16810ea25bc..0524aba4c32 100644
--- a/ndb/src/mgmsrv/Services.cpp
+++ b/ndb/src/mgmsrv/Services.cpp
@@ -840,9 +840,8 @@ MgmApiSession::setClusterLogLevel(Parser<MgmApiSession>::Context &,
   m_mgmsrv.m_event_listner.unlock();
 
   {
-    LogLevel ll;
-    ll.setLogLevel(category,level);
-    m_mgmsrv.m_event_listner.update_max_log_level(ll);
+    LogLevel tmp;
+    m_mgmsrv.m_event_listner.update_max_log_level(tmp);
   }
 
   m_output->println(reply);
@@ -1370,21 +1369,23 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData, NodeId nodeId)
 void
 Ndb_mgmd_event_service::update_max_log_level(const LogLevel &log_level)
 {
-  LogLevel tmp= m_logLevel;
-  tmp.set_max(log_level);
+  LogLevel tmp = log_level;
+  m_clients.lock();
+  for(int i = m_clients.size() - 1; i >= 0; i--)
+    tmp.set_max(m_clients[i].m_logLevel);
+  m_clients.unlock();
   update_log_level(tmp);
 }
 
 void
 Ndb_mgmd_event_service::update_log_level(const LogLevel &tmp)
 {
-  if(!(tmp == m_logLevel)){
-    m_logLevel = tmp;
-    EventSubscribeReq req;
-    req = tmp;
-    req.blockRef = 0;
-    m_mgmsrv->m_log_level_requests.push_back(req);
-  }
+  m_logLevel = tmp;
+  EventSubscribeReq req;
+  req = tmp;
+  // send update to all nodes
+  req.blockRef = 0;
+  m_mgmsrv->m_log_level_requests.push_back(req);
 }
 
 void
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index a3e2e69ce14..bd50440b3c0 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -2267,7 +2267,7 @@ NdbDictionaryImpl::dropIndex(const char * indexName,
     m_error.code = 4243;
     return -1;
   }
-  int ret = dropIndex(*idx); //, tableName);
+  int ret = dropIndex(*idx);
   // If index stored in cache is incompatible with the one in the kernel
   // we must clear the cache and try again
   if (ret == INCOMPATIBLE_VERSION) {
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.hpp b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
index b0b7adf25f3..bc9894497f8 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.hpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
@@ -395,7 +395,6 @@ public:
   int createIndex(NdbIndexImpl &ix);
   int dropIndex(const char * indexName, 
 		const char * tableName);
-  //  int dropIndex(NdbIndexImpl &, const char * tableName);
   int dropIndex(NdbIndexImpl &);
   NdbTableImpl * getIndexTable(NdbIndexImpl * index, 
 			       NdbTableImpl * table);
diff --git a/ndb/src/ndbapi/NdbScanOperation.cpp b/ndb/src/ndbapi/NdbScanOperation.cpp
index 869704c7bb3..7d0712e117d 100644
--- a/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/ndb/src/ndbapi/NdbScanOperation.cpp
@@ -160,7 +160,7 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
     return -1;
   }
 
-  m_keyInfo = lockExcl ? 1 : 0;
+  m_keyInfo = ((scan_flags & SF_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;
diff --git a/ndb/src/ndbapi/ndberror.c b/ndb/src/ndbapi/ndberror.c
index 5ca8ad7be60..a2b96b32630 100644
--- a/ndb/src/ndbapi/ndberror.c
+++ b/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"},