Merge perch.ndb.mysql.com:/home/jonas/src/51-work

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new-ndb

storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -2479,8 +2479,16 @@ Dblqh::execREMOVE_MARKER_ORD(Signal* signal)
   
   CommitAckMarkerPtr removedPtr;
   m_commitAckMarkerHash.remove(removedPtr, key);
+#if defined VM_TRACE || defined ERROR_INSERT
   ndbrequire(removedPtr.i != RNIL);
   m_commitAckMarkerPool.release(removedPtr);
+#else
+  if (removedPtr.i != RNIL)
+  {
+    jam();
+    m_commitAckMarkerPool.release(removedPtr);
+  }
+#endif
 #ifdef MARKER_TRACE
   ndbout_c("Rem marker[%.8x %.8x]", key.transid1, key.transid2);
 #endif
@@ -3138,6 +3146,8 @@ void Dblqh::lqhAttrinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length)
 {
   TcConnectionrec * const regTcPtr = tcConnectptr.p;
   if (regTcPtr->operation != ZREAD) {
+    if (regTcPtr->operation != ZDELETE)
+    {
       if (regTcPtr->opExec != 1) {
 	if (saveTupattrbuf(signal, dataPtr, length) == ZOK) {
 	  ;
@@ -3152,6 +3162,7 @@ void Dblqh::lqhAttrinfoLab(Signal* signal, Uint32* dataPtr, Uint32 length)
 	}//if
       }//if
     }//if
+  }
   c_tup->receive_attrinfo(signal, regTcPtr->tupConnectrec, dataPtr, length);
 }//Dblqh::lqhAttrinfoLab()
 
@@ -3405,7 +3416,7 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
     markerPtr.p->tcNodeId = tcNodeId;
     
     CommitAckMarkerPtr tmp;
-#ifdef VM_TRACE
+#if defined VM_TRACE || defined ERROR_INSERT
 #ifdef MARKER_TRACE
     ndbout_c("Add marker[%.8x %.8x]", markerPtr.p->transid1, markerPtr.p->transid2);
 #endif
@@ -9629,7 +9640,7 @@ Uint32 Dblqh::initScanrec(const ScanFragReq* scanFragReq)
   active.add(scanptr);
   if(scanptr.p->scanKeyinfoFlag){
     jam();
-#ifdef VM_TRACE
+#if defined VM_TRACE || defined ERROR_INSERT
     ScanRecordPtr tmp;
     ndbrequire(!c_scanTakeOverHash.find(tmp, * scanptr.p));
 #endif
@@ -9753,7 +9764,7 @@ void Dblqh::finishScanrec(Signal* signal)
   scans.add(restart);
   if(restart.p->scanKeyinfoFlag){
     jam();
-#ifdef VM_TRACE
+#if defined VM_TRACE || defined ERROR_INSERT
     ScanRecordPtr tmp;
     ndbrequire(!c_scanTakeOverHash.find(tmp, * restart.p));
 #endif

storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -2824,6 +2824,12 @@ void Dbtc::execTCKEYREQ(Signal* signal)
         tmp.p->apiNodeId     = refToNode(regApiPtr->ndbapiBlockref);
         tmp.p->apiConnectPtr = TapiIndex;
         tmp.p->noOfLqhs      = 0;
+#if defined VM_TRACE || defined ERROR_INSERT
+	{
+	  CommitAckMarkerPtr check;
+	  ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+	}
+#endif
         m_commitAckMarkerHash.add(tmp);
       }
     }
@@ -8114,6 +8120,13 @@ void Dbtc::initApiConnectFail(Signal* signal)
     tmp.p->noOfLqhs      = 1;
     tmp.p->lqhNodeId[0]  = tnodeid;
     tmp.p->apiConnectPtr = apiConnectptr.i;
+
+#if defined VM_TRACE || defined ERROR_INSERT
+    {
+      CommitAckMarkerPtr check;
+      ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+    }
+#endif
     m_commitAckMarkerHash.add(tmp);
   } 
 }//Dbtc::initApiConnectFail()
@@ -8270,6 +8283,12 @@ void Dbtc::updateApiStateFail(Signal* signal)
       tmp.p->noOfLqhs      = 1;
       tmp.p->lqhNodeId[0]  = tnodeid;
       tmp.p->apiConnectPtr = apiConnectptr.i;
+#if defined VM_TRACE || defined ERROR_INSERT
+      {
+	CommitAckMarkerPtr check;
+	ndbrequire(!m_commitAckMarkerHash.find(check, *tmp.p));
+      }
+#endif
       m_commitAckMarkerHash.add(tmp);
     } else {
       jam();

storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -1735,7 +1735,8 @@ private:
                       Operationrec* regOperPtr,
                       Fragrecord* regFragPtr,
                       Tablerec* regTabPtr,
-                      KeyReqStruct* req_struct);
+                      KeyReqStruct* req_struct,
+		      bool disk);
 
 //------------------------------------------------------------------
 //------------------------------------------------------------------

storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp

@@ -814,7 +814,9 @@ void Dbtup::execTUPKEYREQ(Signal* signal)
      {
        jam();
        if (handleDeleteReq(signal, regOperPtr,
-			   regFragPtr, regTabPtr, &req_struct) == -1) {
+			   regFragPtr, regTabPtr, 
+			   &req_struct,
+			   disk_page != RNIL) == -1) {
 	 return;
        }
        /*
@@ -1458,7 +1460,8 @@ int Dbtup::handleDeleteReq(Signal* signal,
                            Operationrec* regOperPtr,
                            Fragrecord* regFragPtr,
                            Tablerec* regTabPtr,
-                           KeyReqStruct *req_struct)
+                           KeyReqStruct *req_struct,
+			   bool disk)
 {
   // delete must set but not increment tupVersion
   if (!regOperPtr->is_first_operation())
@@ -1511,7 +1514,10 @@ int Dbtup::handleDeleteReq(Signal* signal,
     return 0;
   }
 
+  if (setup_read(req_struct, regOperPtr, regFragPtr, regTabPtr, disk))
+  {
     return handleReadReq(signal, regOperPtr, regTabPtr, req_struct);
+  }
 
 error:
   tupkeyErrorLab(signal);

storage/ndb/src/kernel/blocks/qmgr/QmgrMain.cpp

@@ -3525,8 +3525,10 @@ void Qmgr::execCOMMIT_FAILREQ(Signal* signal)
       nodePtr.p->phase = ZFAIL_CLOSING;
       nodePtr.p->failState = WAITING_FOR_NDB_FAILCONF;
       setNodeInfo(nodePtr.i).m_heartbeat_cnt= 0;
+      setNodeInfo(nodePtr.i).m_version = 0;
       c_clusterNodes.clear(nodePtr.i);
     }//for
+    recompute_version_info(NodeInfo::DB);
     /*----------------------------------------------------------------------*/
     /*       WE INFORM THE API'S WE HAVE CONNECTED ABOUT THE FAILED NODES.  */
     /*----------------------------------------------------------------------*/

storage/ndb/test/ndbapi/testBasic.cpp

@@ -1272,6 +1272,52 @@ runBug25090(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_OK;
 }
 
+int
+runDeleteRead(NDBT_Context* ctx, NDBT_Step* step){
+  
+  Ndb* pNdb = GETNDB(step);
+
+  const NdbDictionary::Table* tab = ctx->getTab();
+  NDBT_ResultRow row(*ctx->getTab());
+  HugoTransactions tmp(*ctx->getTab());
+
+  int a;
+  int loops = ctx->getNumLoops();
+  const int rows = ctx->getNumRecords();
+  
+  while (loops--)
+  {
+    NdbTransaction* pTrans = pNdb->startTransaction();
+    NdbOperation* pOp = pTrans->getNdbOperation(tab->getName());
+    pOp->deleteTuple();
+    for(a = 0; a<tab->getNoOfColumns(); a++)
+    {
+      if (tab->getColumn(a)->getPrimaryKey() == true)
+      {
+	if(tmp.equalForAttr(pOp, a, 0) != 0)
+	{
+	  ERR(pTrans->getNdbError());
+	  return NDBT_FAILED;
+	}
+      }
+    }
+    
+    // Define attributes to read  
+    for(a = 0; a<tab->getNoOfColumns(); a++)
+    {
+      if((row.attributeStore(a) = pOp->getValue(tab->getColumn(a)->getName())) == 0) {
+	ERR(pTrans->getNdbError());
+	return NDBT_FAILED;
+      }
+    }
+
+    pTrans->execute(Commit);
+    pTrans->close();
+  }
+  
+  return NDBT_OK;
+}
+
 NDBT_TESTSUITE(testBasic);
 TESTCASE("PkInsert", 
 	 "Verify that we can insert and delete from this table using PK"
@@ -1542,6 +1588,12 @@ TESTCASE("Bug25090",
 	 "Verify what happens when we fill the db" ){
   STEP(runBug25090);
 }
+TESTCASE("DeleteRead", 
+	 "Verify Delete+Read" ){
+  INITIALIZER(runLoadTable);
+  INITIALIZER(runDeleteRead);
+  FINALIZER(runClearTable2);
+}
 NDBT_TESTSUITE_END(testBasic);
 
 #if 0

storage/ndb/test/run-test/daily-basic-tests.txt

@@ -79,6 +79,10 @@ max-time: 660
 cmd: testBasic
 args: -n UpdateAndRead 
 
+max-time: 500
+cmd: testBasic
+args: -n DeleteRead
+
 max-time: 500
 cmd: testBasic
 args: -n PkReadAndLocker T6 D1 D2
@@ -461,7 +465,7 @@ max-time: 500
 cmd: testScan
 args: -n Bug24447 T1
 
-max-time: 500
+max-time: 1000
 cmd: testScan
 args: -n ScanVariants