diff --git a/mysql-test/suite/ndb/t/disabled.def b/mysql-test/suite/ndb/t/disabled.def
index 9c2dc80d5ee..f876039a042 100644
--- a/mysql-test/suite/ndb/t/disabled.def
+++ b/mysql-test/suite/ndb/t/disabled.def
@@ -9,9 +9,6 @@
 #  Do not use any TAB characters for whitespace.
 #
 ##############################################################################
-ndb_dd_sql_features      : Bug#29102 ndb_dd_sql_features fails in pushbuild
-ndb_load                 : BUG#17233 2006-05-04 tomas failed load data from infile causes mysqld dbug_assert, binlog not flushed
-
 partition_03ndb          : BUG#16385 2006-03-24 mikael Partitions: crash when updating a range partitioned NDB table
 
 ndb_partition_error2	 : HF is not sure if the test can work as internded on all the platforms
diff --git a/storage/ndb/include/kernel/GlobalSignalNumbers.h b/storage/ndb/include/kernel/GlobalSignalNumbers.h
index aa0596f102a..9653c20260f 100644
--- a/storage/ndb/include/kernel/GlobalSignalNumbers.h
+++ b/storage/ndb/include/kernel/GlobalSignalNumbers.h
@@ -195,9 +195,11 @@ extern const GlobalSignalNumber NO_OF_SIGNAL_NAMES;
 /* 132 not unused */
 /* 133 not unused */
 #define GSN_CM_HEARTBEAT                134 /* distr. */
-/* 135 unused */
-/* 136 unused */
-/* 137 unused */
+
+#define GSN_PREPARE_COPY_FRAG_REQ       135
+#define GSN_PREPARE_COPY_FRAG_REF       136
+#define GSN_PREPARE_COPY_FRAG_CONF      137
+
 #define GSN_CM_NODEINFOCONF             138 /* distr. */
 #define GSN_CM_NODEINFOREF              139 /* distr. */
 #define GSN_CM_NODEINFOREQ              140 /* distr. */
diff --git a/storage/ndb/include/kernel/signaldata/AccScan.hpp b/storage/ndb/include/kernel/signaldata/AccScan.hpp
index 73d69825069..a0aa38c8d8e 100644
--- a/storage/ndb/include/kernel/signaldata/AccScan.hpp
+++ b/storage/ndb/include/kernel/signaldata/AccScan.hpp
@@ -49,6 +49,7 @@ private:
     Uint32 savePointId;
     Uint32 gci;
   };
+  Uint32 maxPage;
 
   /**
    * Previously there where also a scan type
diff --git a/storage/ndb/include/kernel/signaldata/CopyFrag.hpp b/storage/ndb/include/kernel/signaldata/CopyFrag.hpp
index 06dd4070264..d985358dce4 100644
--- a/storage/ndb/include/kernel/signaldata/CopyFrag.hpp
+++ b/storage/ndb/include/kernel/signaldata/CopyFrag.hpp
@@ -29,7 +29,7 @@ class CopyFragReq {
    */
   friend class Dblqh;
 public:
-  STATIC_CONST( SignalLength = 9 );
+  STATIC_CONST( SignalLength = 10 );
 
 private:
   Uint32 userPtr;
@@ -42,6 +42,7 @@ private:
   Uint32 gci;
   Uint32 nodeCount;
   Uint32 nodeList[1];
+  //Uint32 maxPage; is stored in nodeList[nodeCount]
 };
 
 class CopyFragConf {
@@ -95,4 +96,42 @@ struct UpdateFragDistKeyOrd
   STATIC_CONST( SignalLength = 3 );
 };
 
+struct PrepareCopyFragReq
+{
+  STATIC_CONST( SignalLength = 6 );
+
+  Uint32 senderRef;
+  Uint32 senderData;
+  Uint32 tableId;
+  Uint32 fragId;
+  Uint32 copyNodeId;
+  Uint32 startingNodeId;
+};
+
+struct PrepareCopyFragRef
+{
+  Uint32 senderRef;
+  Uint32 senderData;
+  Uint32 tableId;
+  Uint32 fragId;
+  Uint32 copyNodeId;
+  Uint32 startingNodeId;
+  Uint32 errorCode;
+
+  STATIC_CONST( SignalLength = 7 );
+};
+
+struct PrepareCopyFragConf
+{
+  STATIC_CONST( SignalLength = 7 );
+
+  Uint32 senderRef;
+  Uint32 senderData;
+  Uint32 tableId;
+  Uint32 fragId;
+  Uint32 copyNodeId;
+  Uint32 startingNodeId;
+  Uint32 maxPageNo;
+};
+
 #endif
diff --git a/storage/ndb/include/kernel/signaldata/TcKeyConf.hpp b/storage/ndb/include/kernel/signaldata/TcKeyConf.hpp
index b8562875ef5..fd8932c3c87 100644
--- a/storage/ndb/include/kernel/signaldata/TcKeyConf.hpp
+++ b/storage/ndb/include/kernel/signaldata/TcKeyConf.hpp
@@ -46,7 +46,7 @@ public:
    */
   STATIC_CONST( StaticLength = 5 );
   STATIC_CONST( OperationLength = 2 );
-  STATIC_CONST( SimpleReadBit = (((Uint32)1) << 31) );
+  STATIC_CONST( DirtyReadBit = (((Uint32)1) << 31) );
   
 private:
 
diff --git a/storage/ndb/include/ndb_version.h.in b/storage/ndb/include/ndb_version.h.in
index 5405ad4d7aa..6a479433b3b 100644
--- a/storage/ndb/include/ndb_version.h.in
+++ b/storage/ndb/include/ndb_version.h.in
@@ -88,5 +88,52 @@ Uint32 ndbGetOwnVersion();
 
 #define NDBD_NODE_VERSION_REP NDB_MAKE_VERSION(6,1,1)
 
+#define NDBD_PREPARE_COPY_FRAG_VERSION NDB_MAKE_VERSION(6,2,1)
+#define NDBD_PREPARE_COPY_FRAG_V2_51 NDB_MAKE_VERSION(5,1,23)
+#define NDBD_PREPARE_COPY_FRAG_V2_62 NDB_MAKE_VERSION(6,2,8)
+#define NDBD_PREPARE_COPY_FRAG_V2_63 NDB_MAKE_VERSION(6,3,6)
+
+/**
+ * 0 = NO PREP COPY FRAG SUPPORT
+ * 1 = NO MAX PAGE SUPPORT
+ * 2 = LATEST VERSION
+ */
+static
+inline
+int
+ndb_check_prep_copy_frag_version(Uint32 version)
+{	
+  if (version == NDB_VERSION_D)
+    return 2;
+
+  const Uint32 major = (version >> 16) & 0xFF;
+  const Uint32 minor = (version >>  8) & 0xFF;
+  if (major >= 6)
+  {
+    if (minor == 2)
+    {
+      if (version >= NDBD_PREPARE_COPY_FRAG_V2_62)
+        return 2;
+      if (version >= NDBD_PREPARE_COPY_FRAG_VERSION)
+        return 1;
+      return 0;
+    }
+    else if (minor == 3)
+    {
+      if (version >= NDBD_PREPARE_COPY_FRAG_V2_63)
+        return 2;
+      return 1;
+    }
+    return 2;
+  }
+  else if (major == 5 && minor == 1)
+  {
+    if (version >= NDBD_PREPARE_COPY_FRAG_V2_51)
+      return 2;
+  }
+    
+  return 0;
+}
+
 #endif
  
diff --git a/storage/ndb/include/ndbapi/NdbOperation.hpp b/storage/ndb/include/ndbapi/NdbOperation.hpp
index 2f899d5b4bd..78dbadfd7ab 100644
--- a/storage/ndb/include/ndbapi/NdbOperation.hpp
+++ b/storage/ndb/include/ndbapi/NdbOperation.hpp
@@ -93,8 +93,9 @@ public:
     ,LM_CommittedRead       ///< Ignore locks, read last committed value
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
     = 2,
-    LM_Dirty = 2
+    LM_Dirty = 2,
 #endif
+    LM_SimpleRead = 3       ///< Read with shared lock, but release lock directly
   };
 
   /**
diff --git a/storage/ndb/src/common/debugger/signaldata/SignalNames.cpp b/storage/ndb/src/common/debugger/signaldata/SignalNames.cpp
index 0d31cd5de7f..b4221cbec8e 100644
--- a/storage/ndb/src/common/debugger/signaldata/SignalNames.cpp
+++ b/storage/ndb/src/common/debugger/signaldata/SignalNames.cpp
@@ -640,5 +640,9 @@ const GsnName SignalNames [] = {
 
   ,{ GSN_ROUTE_ORD, "ROUTE_ORD" }
   ,{ GSN_NODE_VERSION_REP, "NODE_VERSION_REP" }
+
+  ,{ GSN_PREPARE_COPY_FRAG_REQ,   "PREPARE_COPY_FRAG_REQ" }
+  ,{ GSN_PREPARE_COPY_FRAG_REF,   "PREPARE_COPY_FRAG_REF" }
+  ,{ GSN_PREPARE_COPY_FRAG_CONF,  "PREPARE_COPY_FRAG_CONF" }
 };
 const unsigned short NO_OF_SIGNAL_NAMES = sizeof(SignalNames)/sizeof(GsnName);
diff --git a/storage/ndb/src/common/debugger/signaldata/TcKeyConf.cpp b/storage/ndb/src/common/debugger/signaldata/TcKeyConf.cpp
index 65589f8cd6e..377863f9446 100644
--- a/storage/ndb/src/common/debugger/signaldata/TcKeyConf.cpp
+++ b/storage/ndb/src/common/debugger/signaldata/TcKeyConf.cpp
@@ -51,11 +51,11 @@ printTCKEYCONF(FILE * output, const Uint32 * theData, Uint32 len, Uint16 receive
 	    (TcKeyConf::getMarkerFlag(confInfo) == 0)?"false":"true");
     fprintf(output, "Operations:\n");
     for(i = 0; i < noOfOp; i++) {
-      if(sig->operations[i].attrInfoLen > TcKeyConf::SimpleReadBit)
+      if(sig->operations[i].attrInfoLen > TcKeyConf::DirtyReadBit)
 	fprintf(output,
 		" apiOperationPtr: H'%.8x, simplereadnode: %u\n",
 		sig->operations[i].apiOperationPtr,
-		sig->operations[i].attrInfoLen & (~TcKeyConf::SimpleReadBit));
+		sig->operations[i].attrInfoLen & (~TcKeyConf::DirtyReadBit));
       else
 	fprintf(output,
 		" apiOperationPtr: H'%.8x, attrInfoLen: %u\n",
diff --git a/storage/ndb/src/kernel/blocks/ERROR_codes.txt b/storage/ndb/src/kernel/blocks/ERROR_codes.txt
index 3282cd41fa0..4d4d4fcafc4 100644
--- a/storage/ndb/src/kernel/blocks/ERROR_codes.txt
+++ b/storage/ndb/src/kernel/blocks/ERROR_codes.txt
@@ -3,7 +3,7 @@ Next NDBCNTR 1002
 Next NDBFS 2000
 Next DBACC 3002
 Next DBTUP 4029
-Next DBLQH 5045
+Next DBLQH 5047
 Next DBDICT 6008
 Next DBDIH 7193
 Next DBTC 8054
@@ -11,7 +11,7 @@ Next CMVMI 9000
 Next BACKUP 10038
 Next DBUTIL 11002
 Next DBTUX 12008
-Next SUMA 13001
+Next SUMA 13034
 
 TESTING NODE FAILURE, ARBITRATION
 ---------------------------------
@@ -186,6 +186,8 @@ handling in DBTC to ensure that node failures are also well handled in
 time-out handling. They can also be used to test multiple node failure
 handling.
 
+5045: Crash in PREPARE_COPY_FRAG_REQ
+5046: Crash if LQHKEYREQ (NrCopy) comes when frag-state is incorrect
 
 ERROR CODES FOR TESTING TIME-OUT HANDLING IN DBLQH
 -------------------------------------------------
diff --git a/storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp b/storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
index 1177500bc27..21826df28f9 100644
--- a/storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
+++ b/storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
@@ -545,7 +545,8 @@ public:
       TO_WAIT_ENDING = 21,
       ENDING = 22,
       
-      STARTING_LOCAL_FRAGMENTS = 24
+      STARTING_LOCAL_FRAGMENTS = 24,
+      PREPARE_COPY = 25
     };
     enum ToSlaveStatus {
       TO_SLAVE_IDLE = 0,
@@ -556,6 +557,7 @@ public:
       TO_SLAVE_COPY_COMPLETED = 5
     };
     Uint32 startGci;
+    Uint32 maxPage;
     Uint32 toCopyNode;
     Uint32 toCurrentFragid;
     Uint32 toCurrentReplica;
@@ -672,6 +674,8 @@ private:
   void execNODE_FAILREP(Signal *);
   void execCOPY_FRAGCONF(Signal *);
   void execCOPY_FRAGREF(Signal *);
+  void execPREPARE_COPY_FRAG_REF(Signal*);
+  void execPREPARE_COPY_FRAG_CONF(Signal*);
   void execDIADDTABREQ(Signal *);
   void execDIGETNODESREQ(Signal *);
   void execDIRELEASEREQ(Signal *);
@@ -1114,6 +1118,7 @@ private:
   void sendStartTo(Signal *, Uint32 takeOverPtr);
   void startNextCopyFragment(Signal *, Uint32 takeOverPtr);
   void toCopyFragLab(Signal *, Uint32 takeOverPtr);
+  void toStartCopyFrag(Signal *, TakeOverRecordPtr);
   void startHsAddFragConfLab(Signal *);
   void prepareSendCreateFragReq(Signal *, Uint32 takeOverPtr);
   void sendUpdateTo(Signal *, Uint32 takeOverPtr, Uint32 updateState);
diff --git a/storage/ndb/src/kernel/blocks/dbdih/DbdihInit.cpp b/storage/ndb/src/kernel/blocks/dbdih/DbdihInit.cpp
index aff31d625f4..6ce281434c2 100644
--- a/storage/ndb/src/kernel/blocks/dbdih/DbdihInit.cpp
+++ b/storage/ndb/src/kernel/blocks/dbdih/DbdihInit.cpp
@@ -259,6 +259,11 @@ Dbdih::Dbdih(Block_context& ctx):
   
   addRecSignal(GSN_START_FRAGREF,
 	       &Dbdih::execSTART_FRAGREF);
+
+  addRecSignal(GSN_PREPARE_COPY_FRAG_REF,
+	       &Dbdih::execPREPARE_COPY_FRAG_REF);
+  addRecSignal(GSN_PREPARE_COPY_FRAG_CONF,
+	       &Dbdih::execPREPARE_COPY_FRAG_CONF);
   
   apiConnectRecord = 0;
   connectRecord = 0;
diff --git a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
index dc35e6fba41..5403ac5cc38 100644
--- a/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
@@ -3155,6 +3155,94 @@ void Dbdih::toCopyFragLab(Signal* signal,
   TakeOverRecordPtr takeOverPtr;
   RETURN_IF_TAKE_OVER_INTERRUPTED(takeOverPtrI, takeOverPtr);
 
+  /**
+   * Inform starting node that TakeOver is about to start
+   */
+  Uint32 nodeId = takeOverPtr.p->toStartingNode;
+
+  Uint32 version = getNodeInfo(nodeId).m_version;
+  if (ndb_check_prep_copy_frag_version(version))
+  {
+    jam();
+    TabRecordPtr tabPtr;
+    tabPtr.i = takeOverPtr.p->toCurrentTabref;
+    ptrCheckGuard(tabPtr, ctabFileSize, tabRecord);
+
+    FragmentstorePtr fragPtr;
+    getFragstore(tabPtr.p, takeOverPtr.p->toCurrentFragid, fragPtr);
+    Uint32 nodes[MAX_REPLICAS];
+    extractNodeInfo(fragPtr.p, nodes);
+    
+    PrepareCopyFragReq* req= (PrepareCopyFragReq*)signal->getDataPtrSend();
+    req->senderRef = reference();
+    req->senderData = takeOverPtrI;
+    req->tableId = takeOverPtr.p->toCurrentTabref;
+    req->fragId = takeOverPtr.p->toCurrentFragid;
+    req->copyNodeId = nodes[0]; // Src
+    req->startingNodeId = takeOverPtr.p->toStartingNode; // Dst
+    Uint32 ref = calcLqhBlockRef(takeOverPtr.p->toStartingNode);
+    
+    sendSignal(ref, GSN_PREPARE_COPY_FRAG_REQ, signal, 
+	       PrepareCopyFragReq::SignalLength, JBB);
+    
+    takeOverPtr.p->toMasterStatus = TakeOverRecord::PREPARE_COPY;
+    return;
+  }
+
+  takeOverPtr.p->maxPage = RNIL;
+  toStartCopyFrag(signal, takeOverPtr);
+}
+
+void
+Dbdih::execPREPARE_COPY_FRAG_REF(Signal* signal)
+{
+  jamEntry();
+  PrepareCopyFragRef ref = *(PrepareCopyFragRef*)signal->getDataPtr();
+
+  TakeOverRecordPtr takeOverPtr;
+  RETURN_IF_TAKE_OVER_INTERRUPTED(ref.senderData, takeOverPtr);
+
+  ndbrequire(takeOverPtr.p->toMasterStatus == TakeOverRecord::PREPARE_COPY);
+  
+  /**
+   * Treat this as copy frag ref
+   */
+  CopyFragRef * cfref = (CopyFragRef*)signal->getDataPtrSend();
+  cfref->userPtr = ref.senderData;
+  cfref->startingNodeId = ref.startingNodeId;
+  cfref->errorCode = ref.errorCode;
+  cfref->tableId = ref.tableId;
+  cfref->fragId = ref.fragId;
+  cfref->sendingNodeId = ref.copyNodeId;
+  takeOverPtr.p->toMasterStatus = TakeOverRecord::COPY_FRAG;
+  execCOPY_FRAGREF(signal);
+}
+
+void
+Dbdih::execPREPARE_COPY_FRAG_CONF(Signal* signal)
+{
+  PrepareCopyFragConf conf = *(PrepareCopyFragConf*)signal->getDataPtr();
+
+  TakeOverRecordPtr takeOverPtr;
+  RETURN_IF_TAKE_OVER_INTERRUPTED(conf.senderData, takeOverPtr);
+
+  Uint32 version = getNodeInfo(refToNode(conf.senderRef)).m_version;
+  if (ndb_check_prep_copy_frag_version(version) >= 2)
+  {
+    jam();
+    takeOverPtr.p->maxPage = conf.maxPageNo;
+  }
+  else
+  {
+    jam();
+    takeOverPtr.p->maxPage = RNIL;
+  }
+  toStartCopyFrag(signal, takeOverPtr);
+}
+
+void
+Dbdih::toStartCopyFrag(Signal* signal, TakeOverRecordPtr takeOverPtr)
+{
   CreateReplicaRecordPtr createReplicaPtr;
   createReplicaPtr.i = 0;
   ptrAss(createReplicaPtr, createReplicaRecord);
@@ -3178,8 +3266,8 @@ void Dbdih::toCopyFragLab(Signal* signal,
   createReplicaPtr.p->hotSpareUse = true;
   createReplicaPtr.p->dataNodeId = takeOverPtr.p->toStartingNode;
 
-  prepareSendCreateFragReq(signal, takeOverPtrI);
-}//Dbdih::toCopyFragLab()
+  prepareSendCreateFragReq(signal, takeOverPtr.i);
+}//Dbdih::toStartCopy()
 
 void Dbdih::prepareSendCreateFragReq(Signal* signal, Uint32 takeOverPtrI)
 {
@@ -3412,10 +3500,12 @@ void Dbdih::execCREATE_FRAGCONF(Signal* signal)
     copyFragReq->schemaVersion = tabPtr.p->schemaVersion;
     copyFragReq->distributionKey = fragPtr.p->distributionKey;
     copyFragReq->gci = gci;
-    copyFragReq->nodeCount = extractNodeInfo(fragPtr.p, 
-					     copyFragReq->nodeList);
+    Uint32 len = copyFragReq->nodeCount = 
+      extractNodeInfo(fragPtr.p, 
+                      copyFragReq->nodeList);
+    copyFragReq->nodeList[len] = takeOverPtr.p->maxPage;
     sendSignal(ref, GSN_COPY_FRAGREQ, signal, 
-	       CopyFragReq::SignalLength +  copyFragReq->nodeCount, JBB);
+               CopyFragReq::SignalLength + len, JBB);
   } else {
     ndbrequire(takeOverPtr.p->toMasterStatus == TakeOverRecord::COMMIT_CREATE);
     jam();
@@ -4576,13 +4666,22 @@ void Dbdih::checkTakeOverInMasterStartNodeFailure(Signal* signal,
     ok = true;
     jam();
     //-----------------------------------------------------------------------
-    // The starting node will discover the problem. We will receive either
+    // The copying node will discover the problem. We will receive either
     // COPY_FRAGREQ or COPY_FRAGCONF and then we can release the take over
     // record and end the process. If the copying node should also die then
     // we will try to send prepare create fragment and will then discover
     // that the starting node has failed.
     //-----------------------------------------------------------------------
     break;
+  case TakeOverRecord::PREPARE_COPY:
+    ok = true;
+    jam();
+    /**
+     * We're waiting for the starting node...which just died...
+     *  endTakeOver
+     */
+    endTakeOver(takeOverPtr.i);
+    break;
   case TakeOverRecord::COPY_ACTIVE:
     ok = true;
     jam();
diff --git a/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp b/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
index 6f8e5569831..95cad98b81c 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
@@ -2025,7 +2025,6 @@ public:
     Uint8 reclenAiLqhkey;
     Uint8 m_offset_current_keybuf;
     Uint8 replicaType;
-    Uint8 simpleRead;
     Uint8 seqNoReplica;
     Uint8 tcNodeFailrec;
     Uint8 m_disk_table;
@@ -2145,6 +2144,7 @@ private:
   void execSTORED_PROCCONF(Signal* signal);
   void execSTORED_PROCREF(Signal* signal);
   void execCOPY_FRAGREQ(Signal* signal);
+  void execPREPARE_COPY_FRAG_REQ(Signal* signal);
   void execUPDATE_FRAG_DIST_KEY_ORD(Signal*);
   void execCOPY_ACTIVEREQ(Signal* signal);
   void execCOPY_STATEREQ(Signal* signal);
diff --git a/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp b/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
index d6411ee1cb9..db6d201575f 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
@@ -310,6 +310,9 @@ Dblqh::Dblqh(Block_context& ctx):
   addRecSignal(GSN_UPDATE_FRAG_DIST_KEY_ORD, 
 	       &Dblqh::execUPDATE_FRAG_DIST_KEY_ORD);
   
+  addRecSignal(GSN_PREPARE_COPY_FRAG_REQ,
+	       &Dblqh::execPREPARE_COPY_FRAG_REQ);
+  
   initData();
 
 #ifdef VM_TRACE
diff --git a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
index 43d49e791be..e0449e08ddd 100644
--- a/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
@@ -3496,7 +3496,6 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
   regTcPtr->dirtyOp       = LqhKeyReq::getDirtyFlag(Treqinfo);
   regTcPtr->opExec        = LqhKeyReq::getInterpretedFlag(Treqinfo);
   regTcPtr->opSimple      = LqhKeyReq::getSimpleFlag(Treqinfo);
-  regTcPtr->simpleRead    = op == ZREAD && regTcPtr->opSimple;
   regTcPtr->seqNoReplica  = LqhKeyReq::getSeqNoReplica(Treqinfo);
   UintR TreclenAiLqhkey   = LqhKeyReq::getAIInLqhKeyReq(Treqinfo);
   regTcPtr->apiVersionNo  = 0; 
@@ -3513,9 +3512,15 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
     regTcPtr->lockType = 
       op == ZREAD_EX ? ZUPDATE : (Operation_t) op == ZWRITE ? ZINSERT : (Operation_t) op;
   }
+
+  if (regTcPtr->dirtyOp)
+  {
+    ndbrequire(regTcPtr->opSimple);
+  }
   
-  CRASH_INSERTION2(5041, regTcPtr->simpleRead && 
-		   refToNode(signal->senderBlockRef()) != cownNodeid);
+  CRASH_INSERTION2(5041, (op == ZREAD && 
+                          (regTcPtr->opSimple || regTcPtr->dirtyOp) &&
+                          refToNode(signal->senderBlockRef()) != cownNodeid));
   
   regTcPtr->reclenAiLqhkey = TreclenAiLqhkey;
   regTcPtr->currReclenAi = TreclenAiLqhkey;
@@ -3665,6 +3670,7 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
     {
       ndbout_c("fragptr.p->fragStatus: %d",
 	       fragptr.p->fragStatus);
+      CRASH_INSERTION(5046);
     }
     ndbassert(fragptr.p->fragStatus == Fragrecord::ACTIVE_CREATION);
     fragptr.p->m_copy_started_state = Fragrecord::AC_NR_COPY;
@@ -3687,8 +3693,8 @@ void Dblqh::execLQHKEYREQ(Signal* signal)
   Uint8 TdistKey = LqhKeyReq::getDistributionKey(TtotReclenAi);
   if ((tfragDistKey != TdistKey) &&
       (regTcPtr->seqNoReplica == 0) &&
-      (regTcPtr->dirtyOp == ZFALSE) &&
-      (regTcPtr->simpleRead == ZFALSE)) {
+      (regTcPtr->dirtyOp == ZFALSE)) 
+  {
     /* ----------------------------------------------------------------------
      * WE HAVE DIFFERENT OPINION THAN THE DIH THAT STARTED THE TRANSACTION. 
      * THE REASON COULD BE THAT THIS IS AN OLD DISTRIBUTION WHICH IS NO LONGER
@@ -4778,7 +4784,18 @@ void Dblqh::tupkeyConfLab(Signal* signal)
 
   TRACE_OP(regTcPtr, "TUPKEYCONF");
 
-  if (regTcPtr->simpleRead) {
+  if (readLen != 0) 
+  {
+    jam();
+
+    /* SET BIT 15 IN REQINFO */
+    LqhKeyReq::setApplicationAddressFlag(regTcPtr->reqinfo, 1);
+    regTcPtr->readlenAi = readLen;
+  }//if
+
+  if (regTcPtr->operation == ZREAD && 
+      (regTcPtr->opSimple || regTcPtr->dirtyOp))
+  {
     jam();
     /* ----------------------------------------------------------------------
      * THE OPERATION IS A SIMPLE READ. 
@@ -4792,14 +4809,6 @@ void Dblqh::tupkeyConfLab(Signal* signal)
     commitContinueAfterBlockedLab(signal);
     return;
   }//if
-  if (readLen != 0) 
-  {
-    jam();
-
-    /* SET BIT 15 IN REQINFO */
-    LqhKeyReq::setApplicationAddressFlag(regTcPtr->reqinfo, 1);
-    regTcPtr->readlenAi = readLen;
-  }//if
   regTcPtr->totSendlenAi = writeLen;
   ndbrequire(regTcPtr->totSendlenAi == regTcPtr->currTupAiLen);
   
@@ -5178,12 +5187,15 @@ void Dblqh::packLqhkeyreqLab(Signal* signal)
 /*                                                                           */
 /* ------------------------------------------------------------------------- */
     sendLqhkeyconfTc(signal, regTcPtr->tcBlockref);
-    if (regTcPtr->dirtyOp != ZTRUE) {
+    if (! (regTcPtr->dirtyOp || 
+           (regTcPtr->operation == ZREAD && regTcPtr->opSimple)))
+    {
       jam();
       regTcPtr->transactionState = TcConnectionrec::PREPARED;
       releaseOprec(signal);
     } else {
       jam();
+
 /*************************************************************>*/
 /*       DIRTY WRITES ARE USED IN TWO SITUATIONS. THE FIRST    */
 /*       SITUATION IS WHEN THEY ARE USED TO UPDATE COUNTERS AND*/
@@ -6406,8 +6418,8 @@ void Dblqh::commitContinueAfterBlockedLab(Signal* signal)
   Ptr<TcConnectionrec> regTcPtr = tcConnectptr;
   Ptr<Fragrecord> regFragptr = fragptr;
   Uint32 operation = regTcPtr.p->operation;
-  Uint32 simpleRead = regTcPtr.p->simpleRead;
   Uint32 dirtyOp = regTcPtr.p->dirtyOp;
+  Uint32 opSimple = regTcPtr.p->opSimple;
   if (regTcPtr.p->activeCreat != Fragrecord::AC_IGNORED) {
     if (operation != ZREAD) {
       TupCommitReq * const tupCommitReq = 
@@ -6465,20 +6477,29 @@ void Dblqh::commitContinueAfterBlockedLab(Signal* signal)
 	EXECUTE_DIRECT(acc, GSN_ACC_COMMITREQ, signal, 1);
       }
       
-      if (simpleRead) {
+      if (dirtyOp) 
+      {
 	jam();
-/* ------------------------------------------------------------------------- */
-/*THE OPERATION WAS A SIMPLE READ THUS THE COMMIT PHASE IS ONLY NEEDED TO    */
-/*RELEASE THE LOCKS. AT THIS POINT IN THE CODE THE LOCKS ARE RELEASED AND WE */
-/*ARE IN A POSITION TO SEND LQHKEYCONF TO TC. WE WILL ALSO RELEASE ALL       */
-/*RESOURCES BELONGING TO THIS OPERATION SINCE NO MORE WORK WILL BE           */
-/*PERFORMED.                                                                 */
-/* ------------------------------------------------------------------------- */
+        /**
+         * The dirtyRead does not send anything but TRANSID_AI from LDM
+         */
 	fragptr = regFragptr;
 	tcConnectptr = regTcPtr;
 	cleanUp(signal);
 	return;
-      }//if
+      }
+
+      /**
+       * The simpleRead will send a LQHKEYCONF
+       *   but have already released the locks
+       */
+      if (opSimple)
+      {
+	fragptr = regFragptr;
+	tcConnectptr = regTcPtr;
+        packLqhkeyreqLab(signal);
+        return;
+      }
     }
   }//if
   jamEntry();
@@ -7088,7 +7109,7 @@ void Dblqh::abortStateHandlerLab(Signal* signal)
 /* ------------------------------------------------------------------------- */
       return;
     }//if
-    if (regTcPtr->simpleRead) {
+    if (regTcPtr->opSimple) {
       jam();
 /* ------------------------------------------------------------------------- */
 /*A SIMPLE READ IS CURRENTLY RELEASING THE LOCKS OR WAITING FOR ACCESS TO    */
@@ -7356,7 +7377,8 @@ void Dblqh::continueAbortLab(Signal* signal)
 void Dblqh::continueAfterLogAbortWriteLab(Signal* signal) 
 {
   TcConnectionrec * const regTcPtr = tcConnectptr.p;
-  if (regTcPtr->simpleRead) {
+  if (regTcPtr->operation == ZREAD && regTcPtr->dirtyOp)
+  {
     jam();
     TcKeyRef * const tcKeyRef = (TcKeyRef *) signal->getDataPtrSend();
     
@@ -10062,6 +10084,86 @@ Dblqh::calculateHash(Uint32 tableId, const Uint32* src)
   return md5_hash(Tmp, keyLen);
 }//Dblqh::calculateHash()
 
+/**
+ * PREPARE COPY FRAG REQ
+ */
+void
+Dblqh::execPREPARE_COPY_FRAG_REQ(Signal* signal)
+{
+  jamEntry();
+  PrepareCopyFragReq req = *(PrepareCopyFragReq*)signal->getDataPtr();
+
+  CRASH_INSERTION(5045);
+
+  tabptr.i = req.tableId;
+  ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
+
+  Uint32 max_page = RNIL;
+  
+  if (getOwnNodeId() != req.startingNodeId)
+  {
+    jam();
+    /**
+     * This is currently dead code...
+     *   but is provided so we can impl. a better scan+delete on
+     *   starting node wo/ having to change running node
+     */
+    ndbrequire(getOwnNodeId() == req.copyNodeId);
+    c_tup->get_frag_info(req.tableId, req.fragId, &max_page);    
+
+    PrepareCopyFragConf* conf = (PrepareCopyFragConf*)signal->getDataPtrSend();
+    conf->senderData = req.senderData;
+    conf->senderRef = reference();
+    conf->tableId = req.tableId;
+    conf->fragId = req.fragId;
+    conf->copyNodeId = req.copyNodeId;
+    conf->startingNodeId = req.startingNodeId;
+    conf->maxPageNo = max_page;
+    sendSignal(req.senderRef, GSN_PREPARE_COPY_FRAG_CONF,
+               signal, PrepareCopyFragConf::SignalLength, JBB);  
+    
+    return;
+  }
+  
+  if (! DictTabInfo::isOrderedIndex(tabptr.p->tableType))
+  {
+    jam();
+    ndbrequire(getFragmentrec(signal, req.fragId));
+    
+    /**
+     *
+     */
+    if (cstartType == NodeState::ST_SYSTEM_RESTART)
+    {
+      jam();
+      signal->theData[0] = fragptr.p->tabRef;
+      signal->theData[1] = fragptr.p->fragId;
+      sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
+    }
+    
+    
+    /**
+     *
+     */
+    fragptr.p->m_copy_started_state = Fragrecord::AC_IGNORED;
+    fragptr.p->fragStatus = Fragrecord::ACTIVE_CREATION;
+    fragptr.p->logFlag = Fragrecord::STATE_FALSE;
+
+    c_tup->get_frag_info(req.tableId, req.fragId, &max_page);
+  }    
+    
+  PrepareCopyFragConf* conf = (PrepareCopyFragConf*)signal->getDataPtrSend();
+  conf->senderData = req.senderData;
+  conf->senderRef = reference();
+  conf->tableId = req.tableId;
+  conf->fragId = req.fragId;
+  conf->copyNodeId = req.copyNodeId;
+  conf->startingNodeId = req.startingNodeId;
+  conf->maxPageNo = max_page;
+  sendSignal(req.senderRef, GSN_PREPARE_COPY_FRAG_CONF,
+             signal, PrepareCopyFragConf::SignalLength, JBB);  
+}
+
 /* *************************************** */
 /*  COPY_FRAGREQ: Start copying a fragment */
 /* *************************************** */
@@ -10097,6 +10199,13 @@ void Dblqh::execCOPY_FRAGREQ(Signal* signal)
     for (i = 0; i<nodeCount; i++)
       nodemask.set(copyFragReq->nodeList[i]);
   }
+  Uint32 maxPage = copyFragReq->nodeList[nodeCount];
+  Uint32 version = getNodeInfo(refToNode(userRef)).m_version;
+  if (ndb_check_prep_copy_frag_version(version) < 2)
+  {
+    jam();
+    maxPage = RNIL;
+  }
     
   if (DictTabInfo::isOrderedIndex(tabptr.p->tableType)) {
     jam();
@@ -10172,14 +10281,15 @@ void Dblqh::execCOPY_FRAGREQ(Signal* signal)
   req->requestInfo = 0;
   AccScanReq::setLockMode(req->requestInfo, 0);
   AccScanReq::setReadCommittedFlag(req->requestInfo, 0);
-  AccScanReq::setNRScanFlag(req->requestInfo, gci ? 1 : 0);
+  AccScanReq::setNRScanFlag(req->requestInfo, 1);
   AccScanReq::setNoDiskScanFlag(req->requestInfo, 1);
 
   req->transId1 = tcConnectptr.p->transid[0];
   req->transId2 = tcConnectptr.p->transid[1];
   req->savePointId = tcConnectptr.p->savePointId;
+  req->maxPage = maxPage;
   sendSignal(scanptr.p->scanBlockref, GSN_ACC_SCANREQ, signal, 
-	     AccScanReq::SignalLength, JBB);
+	     AccScanReq::SignalLength + 1, JBB);
   
   if (! nodemask.isclear())
   {
@@ -19027,7 +19137,6 @@ Dblqh::execDUMP_STATE_ORD(Signal* signal)
     ndbout << " operation = " << tcRec.p->operation<<endl;
     ndbout << " tcNodeFailrec = " << tcRec.p->tcNodeFailrec
 	   << " seqNoReplica = " << tcRec.p->seqNoReplica
-	   << " simpleRead = " << tcRec.p->simpleRead
 	   << endl;
     ndbout << " replicaType = " << tcRec.p->replicaType
 	   << " reclenAiLqhkey = " << tcRec.p->reclenAiLqhkey
diff --git a/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp b/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
index 75d79ba737f..3d5e52a525d 100644
--- a/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
@@ -786,6 +786,7 @@ public:
     UintR apiConnect;                      /* POINTER TO API CONNECT RECORD */
     UintR nextTcConnect;                   /* NEXT TC RECORD*/
     Uint8 dirtyOp;
+    Uint8 opSimple;   
     Uint8 lastReplicaNo;     /* NUMBER OF THE LAST REPLICA IN THE OPERATION */
     Uint8 noOfNodes;         /* TOTAL NUMBER OF NODES IN OPERATION          */
     Uint8 operation;         /* OPERATION TYPE                              */
@@ -886,13 +887,8 @@ public:
     Uint8  opExec;     
 
     Uint8  unused;
+    Uint8  unused1;
 
-    /** 
-     * IS THE OPERATION A SIMPLE TRANSACTION            
-     * 0 = NO, 1 = YES                                 
-     */
-    Uint8  opSimple;   
-    
     //---------------------------------------------------
     // Second 16 byte cache line in second 64 byte cache
     // line. Diverse use.
@@ -1464,7 +1460,7 @@ private:
   void releaseAttrinfo();
   void releaseGcp(Signal* signal);
   void releaseKeys();
-  void releaseSimpleRead(Signal*, ApiConnectRecordPtr, TcConnectRecord*);
+  void releaseDirtyRead(Signal*, ApiConnectRecordPtr, TcConnectRecord*);
   void releaseDirtyWrite(Signal* signal);
   void releaseTcCon();
   void releaseTcConnectFail(Signal* signal);
@@ -1620,7 +1616,7 @@ private:
   void startphase1x010Lab(Signal* signal);
 
   void lqhKeyConf_checkTransactionState(Signal * signal,
-					ApiConnectRecord * const regApiPtr);
+					Ptr<ApiConnectRecord> regApiPtr);
 
   void checkDropTab(Signal* signal);
 
diff --git a/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp b/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
index 31f6a4eec08..ce20059e663 100644
--- a/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
@@ -2791,9 +2791,9 @@ void Dbtc::execTCKEYREQ(Signal* signal)
   Uint8 TNoDiskFlag         = TcKeyReq::getNoDiskFlag(Treqinfo);
   Uint8 TexecuteFlag        = TexecFlag;
   
-  regCachePtr->opSimple = TSimpleFlag;
-  regCachePtr->opExec   = TInterpretedFlag;
   regTcPtr->dirtyOp  = TDirtyFlag;
+  regTcPtr->opSimple = TSimpleFlag;
+  regCachePtr->opExec   = TInterpretedFlag;
   regCachePtr->distributionKeyIndicator = TDistrKeyFlag;
   regCachePtr->m_no_disk_flag = TNoDiskFlag;
 
@@ -3247,9 +3247,10 @@ void Dbtc::sendlqhkeyreq(Signal* signal,
   LqhKeyReq::setScanTakeOverFlag(tslrAttrLen, regCachePtr->scanTakeOverInd);
 
   Tdata10 = 0;
-  sig0 = regCachePtr->opSimple;
+  sig0 = regTcPtr->opSimple;
   sig1 = regTcPtr->operation;
-  bool simpleRead = (sig1 == ZREAD && sig0 == ZTRUE);
+  sig2 = regTcPtr->dirtyOp;
+  bool dirtyRead = (sig1 == ZREAD && sig2 == ZTRUE);
   LqhKeyReq::setKeyLen(Tdata10, regCachePtr->keylen);
   LqhKeyReq::setLastReplicaNo(Tdata10, regTcPtr->lastReplicaNo);
   if (unlikely(version < NDBD_ROWID_VERSION))
@@ -3262,7 +3263,7 @@ void Dbtc::sendlqhkeyreq(Signal* signal,
   // Indicate Application Reference is present in bit 15
   /* ---------------------------------------------------------------------- */
   LqhKeyReq::setApplicationAddressFlag(Tdata10, 1);
-  LqhKeyReq::setDirtyFlag(Tdata10, regTcPtr->dirtyOp);
+  LqhKeyReq::setDirtyFlag(Tdata10, sig2);
   LqhKeyReq::setInterpretedFlag(Tdata10, regCachePtr->opExec);
   LqhKeyReq::setSimpleFlag(Tdata10, sig0);
   LqhKeyReq::setOperation(Tdata10, sig1);
@@ -3323,7 +3324,7 @@ void Dbtc::sendlqhkeyreq(Signal* signal,
   sig5 = regTcPtr->clientData;
   sig6 = regCachePtr->scanInfo;
 
-  if (! simpleRead)
+  if (! dirtyRead)
   {
     regApiPtr->m_transaction_nodes.set(regTcPtr->tcNodedata[0]);
     regApiPtr->m_transaction_nodes.set(regTcPtr->tcNodedata[1]);
@@ -3396,7 +3397,6 @@ void Dbtc::packLqhkeyreq040Lab(Signal* signal,
                                BlockReference TBRef) 
 {
   TcConnectRecord * const regTcPtr = tcConnectptr.p;
-  CacheRecord * const regCachePtr = cachePtr.p;
 #ifdef ERROR_INSERT
   ApiConnectRecord * const regApiPtr = apiConnectptr.p;
   if (ERROR_INSERTED(8009)) {
@@ -3421,8 +3421,8 @@ void Dbtc::packLqhkeyreq040Lab(Signal* signal,
     if (anAttrBufIndex == RNIL) {
       UintR TtcTimer = ctcTimer;
       UintR Tread = (regTcPtr->operation == ZREAD);
-      UintR Tsimple = (regCachePtr->opSimple == ZTRUE);
-      UintR Tboth = Tread & Tsimple;
+      UintR Tdirty = (regTcPtr->dirtyOp == ZTRUE);
+      UintR Tboth = Tread & Tdirty;
       setApiConTimer(apiConnectptr.i, TtcTimer, __LINE__);
       jam();
       /*--------------------------------------------------------------------
@@ -3431,7 +3431,7 @@ void Dbtc::packLqhkeyreq040Lab(Signal* signal,
       releaseAttrinfo();
       if (Tboth) {
         jam();
-        releaseSimpleRead(signal, apiConnectptr, tcConnectptr.p);
+        releaseDirtyRead(signal, apiConnectptr, tcConnectptr.p);
         return;
       }//if
       regTcPtr->tcConnectstate = OS_OPERATING;
@@ -3491,11 +3491,11 @@ void Dbtc::releaseAttrinfo()
 }//Dbtc::releaseAttrinfo()
 
 /* ========================================================================= */
-/* -------   RELEASE ALL RECORDS CONNECTED TO A SIMPLE OPERATION     ------- */
+/* -------   RELEASE ALL RECORDS CONNECTED TO A DIRTY OPERATION     ------- */
 /* ========================================================================= */
-void Dbtc::releaseSimpleRead(Signal* signal, 
-			     ApiConnectRecordPtr regApiPtr,
-			     TcConnectRecord* regTcPtr) 
+void Dbtc::releaseDirtyRead(Signal* signal, 
+                            ApiConnectRecordPtr regApiPtr,
+                            TcConnectRecord* regTcPtr) 
 {
   Uint32 Ttckeyrec = regApiPtr.p->tckeyrec;
   Uint32 TclientData = regTcPtr->clientData;
@@ -3505,7 +3505,7 @@ void Dbtc::releaseSimpleRead(Signal* signal,
   ConnectionState state = regApiPtr.p->apiConnectstate;
   
   regApiPtr.p->tcSendArray[Ttckeyrec] = TclientData;
-  regApiPtr.p->tcSendArray[Ttckeyrec + 1] = TcKeyConf::SimpleReadBit | Tnode;
+  regApiPtr.p->tcSendArray[Ttckeyrec + 1] = TcKeyConf::DirtyReadBit | Tnode;
   regApiPtr.p->tckeyrec = Ttckeyrec + 2;
   
   unlinkReadyTcCon(signal);
@@ -3535,8 +3535,8 @@ void Dbtc::releaseSimpleRead(Signal* signal,
   /**
    * Emulate LQHKEYCONF
    */
-  lqhKeyConf_checkTransactionState(signal, regApiPtr.p);
-}//Dbtc::releaseSimpleRead()
+  lqhKeyConf_checkTransactionState(signal, regApiPtr);
+}//Dbtc::releaseDirtyRead()
 
 /* ------------------------------------------------------------------------- */
 /* -------        CHECK IF ALL TC CONNECTIONS ARE COMPLETED          ------- */
@@ -3718,12 +3718,13 @@ void Dbtc::execLQHKEYCONF(Signal* signal)
     TCKEY_abort(signal, 29);
     return;
   }//if
-  ApiConnectRecord * const regApiPtr = 
-                            &localApiConnectRecord[TapiConnectptrIndex];
+  Ptr<ApiConnectRecord> regApiPtr;
+  regApiPtr.i = TapiConnectptrIndex;
+  regApiPtr.p = &localApiConnectRecord[TapiConnectptrIndex];
   apiConnectptr.i = TapiConnectptrIndex;
-  apiConnectptr.p = regApiPtr;
-  compare_transid1 = regApiPtr->transid[0] ^ Ttrans1;
-  compare_transid2 = regApiPtr->transid[1] ^ Ttrans2;
+  apiConnectptr.p = regApiPtr.p;
+  compare_transid1 = regApiPtr.p->transid[0] ^ Ttrans1;
+  compare_transid2 = regApiPtr.p->transid[1] ^ Ttrans2;
   compare_transid1 = compare_transid1 | compare_transid2;
   if (compare_transid1 != 0) {
     warningReport(signal, 24);
@@ -3735,25 +3736,25 @@ void Dbtc::execLQHKEYCONF(Signal* signal)
     systemErrorLab(signal, __LINE__);
   }//if
   if (ERROR_INSERTED(8003)) {
-    if (regApiPtr->apiConnectstate == CS_STARTED) {
+    if (regApiPtr.p->apiConnectstate == CS_STARTED) {
       CLEAR_ERROR_INSERT_VALUE;
       return;
     }//if
   }//if
   if (ERROR_INSERTED(8004)) {
-    if (regApiPtr->apiConnectstate == CS_RECEIVING) {
+    if (regApiPtr.p->apiConnectstate == CS_RECEIVING) {
       CLEAR_ERROR_INSERT_VALUE;
       return;
     }//if
   }//if
   if (ERROR_INSERTED(8005)) {
-    if (regApiPtr->apiConnectstate == CS_REC_COMMITTING) {
+    if (regApiPtr.p->apiConnectstate == CS_REC_COMMITTING) {
       CLEAR_ERROR_INSERT_VALUE;
       return;
     }//if
   }//if
   if (ERROR_INSERTED(8006)) {
-    if (regApiPtr->apiConnectstate == CS_START_COMMITTING) {
+    if (regApiPtr.p->apiConnectstate == CS_START_COMMITTING) {
       CLEAR_ERROR_INSERT_VALUE;
       return;
     }//if
@@ -3768,10 +3769,12 @@ void Dbtc::execLQHKEYCONF(Signal* signal)
   regTcPtr->lastLqhNodeId = refToNode(tlastLqhBlockref);
   regTcPtr->noFiredTriggers = noFired;
 
-  UintR Ttckeyrec = (UintR)regApiPtr->tckeyrec;
+  UintR Ttckeyrec = (UintR)regApiPtr.p->tckeyrec;
   UintR TclientData = regTcPtr->clientData;
   UintR TdirtyOp = regTcPtr->dirtyOp;
-  ConnectionState TapiConnectstate = regApiPtr->apiConnectstate;
+  Uint32 TopSimple = regTcPtr->opSimple;
+  Uint32 Toperation = regTcPtr->operation;
+  ConnectionState TapiConnectstate = regApiPtr.p->apiConnectstate;
   if (Ttckeyrec > (ZTCOPCONF_SIZE - 2)) {
     TCKEY_abort(signal, 30);
     return;
@@ -3796,23 +3799,34 @@ void Dbtc::execLQHKEYCONF(Signal* signal)
        * since they will enter execLQHKEYCONF a second time
        * Skip counting internally generated TcKeyReq
        */
-      regApiPtr->tcSendArray[Ttckeyrec] = TclientData;
-      regApiPtr->tcSendArray[Ttckeyrec + 1] = treadlenAi;
-      regApiPtr->tckeyrec = Ttckeyrec + 2;
+      regApiPtr.p->tcSendArray[Ttckeyrec] = TclientData;
+      regApiPtr.p->tcSendArray[Ttckeyrec + 1] = treadlenAi;
+      regApiPtr.p->tckeyrec = Ttckeyrec + 2;
     }//if
   }//if
-  if (TdirtyOp == ZTRUE) {
-    UintR Tlqhkeyreqrec = regApiPtr->lqhkeyreqrec;
+  if (TdirtyOp == ZTRUE) 
+  {
+    UintR Tlqhkeyreqrec = regApiPtr.p->lqhkeyreqrec;
     jam();
     releaseDirtyWrite(signal);
-    regApiPtr->lqhkeyreqrec = Tlqhkeyreqrec - 1;
-  } else {
+    regApiPtr.p->lqhkeyreqrec = Tlqhkeyreqrec - 1;
+  } 
+  else if (Toperation == ZREAD && TopSimple)
+  {
+    UintR Tlqhkeyreqrec = regApiPtr.p->lqhkeyreqrec;
+    jam();
+    unlinkReadyTcCon(signal);
+    releaseTcCon();
+    regApiPtr.p->lqhkeyreqrec = Tlqhkeyreqrec - 1;
+  } 
+  else 
+  {
     jam();
     if (noFired == 0) {
       jam();
       // No triggers to execute
-      UintR Tlqhkeyconfrec = regApiPtr->lqhkeyconfrec;
-      regApiPtr->lqhkeyconfrec = Tlqhkeyconfrec + 1;
+      UintR Tlqhkeyconfrec = regApiPtr.p->lqhkeyconfrec;
+      regApiPtr.p->lqhkeyconfrec = Tlqhkeyconfrec + 1;
       regTcPtr->tcConnectstate = OS_PREPARED;
     }
   }//if
@@ -3842,21 +3856,18 @@ void Dbtc::execLQHKEYCONF(Signal* signal)
     jam();
     if (regTcPtr->isIndexOp) {
       jam();
-      setupIndexOpReturn(regApiPtr, regTcPtr);
+      setupIndexOpReturn(regApiPtr.p, regTcPtr);
     }
     lqhKeyConf_checkTransactionState(signal, regApiPtr);
   } else {
     // We have fired triggers
     jam();
     saveTriggeringOpState(signal, regTcPtr);
-    if (regTcPtr->noReceivedTriggers == noFired) {
-      ApiConnectRecordPtr transPtr;
-      
+    if (regTcPtr->noReceivedTriggers == noFired) 
+    {
       // We have received all data
       jam();
-      transPtr.i = TapiConnectptrIndex;
-      transPtr.p = regApiPtr;
-      executeTriggers(signal, &transPtr);
+      executeTriggers(signal, &regApiPtr);
     }
     // else wait for more trigger data
   }
@@ -3880,7 +3891,7 @@ void Dbtc::setupIndexOpReturn(ApiConnectRecord* regApiPtr,
  */
 void
 Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
-				       ApiConnectRecord * const apiConnectPtrP)
+				       Ptr<ApiConnectRecord> regApiPtr)
 {
 /*---------------------------------------------------------------*/
 /* IF THE COMMIT FLAG IS SET IN SIGNAL TCKEYREQ THEN DBTC HAS TO */
@@ -3891,9 +3902,9 @@ Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
 /* FOR ALL OPERATIONS, AND THEN WAIT FOR THE API TO CONCLUDE THE */
 /* TRANSACTION                                                   */
 /*---------------------------------------------------------------*/
-  ConnectionState TapiConnectstate = apiConnectPtrP->apiConnectstate;
-  UintR Tlqhkeyconfrec = apiConnectPtrP->lqhkeyconfrec;
-  UintR Tlqhkeyreqrec = apiConnectPtrP->lqhkeyreqrec;
+  ConnectionState TapiConnectstate = regApiPtr.p->apiConnectstate;
+  UintR Tlqhkeyconfrec = regApiPtr.p->lqhkeyconfrec;
+  UintR Tlqhkeyreqrec = regApiPtr.p->lqhkeyreqrec;
   int TnoOfOutStanding = Tlqhkeyreqrec - Tlqhkeyconfrec;
 
   switch (TapiConnectstate) {
@@ -3903,11 +3914,11 @@ Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
       diverify010Lab(signal);
       return;
     } else if (TnoOfOutStanding > 0) {
-      if (apiConnectPtrP->tckeyrec == ZTCOPCONF_SIZE) {
+      if (regApiPtr.p->tckeyrec == ZTCOPCONF_SIZE) {
         jam();
         sendtckeyconf(signal, 0);
         return;
-      } else if (apiConnectPtrP->indexOpReturn) {
+      } else if (regApiPtr.p->indexOpReturn) {
 	jam();
         sendtckeyconf(signal, 0);
         return;
@@ -3926,11 +3937,11 @@ Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
       sendtckeyconf(signal, 2);
       return;
     } else {
-      if (apiConnectPtrP->tckeyrec == ZTCOPCONF_SIZE) {
+      if (regApiPtr.p->tckeyrec == ZTCOPCONF_SIZE) {
         jam();
         sendtckeyconf(signal, 0);
         return;
-      } else if (apiConnectPtrP->indexOpReturn) {
+      } else if (regApiPtr.p->indexOpReturn) {
 	jam();
         sendtckeyconf(signal, 0);
         return;
@@ -3940,11 +3951,11 @@ Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
     return;
   case CS_REC_COMMITTING:
     if (TnoOfOutStanding > 0) {
-      if (apiConnectPtrP->tckeyrec == ZTCOPCONF_SIZE) {
+      if (regApiPtr.p->tckeyrec == ZTCOPCONF_SIZE) {
         jam();
         sendtckeyconf(signal, 0);
         return;
-      } else if (apiConnectPtrP->indexOpReturn) {
+      } else if (regApiPtr.p->indexOpReturn) {
         jam();
         sendtckeyconf(signal, 0);
         return;
@@ -3961,7 +3972,7 @@ Dbtc::lqhKeyConf_checkTransactionState(Signal * signal,
 /*       CONSISTING OF DIRTY WRITES AND ALL OF THOSE WERE        */
 /*       COMPLETED. ENSURE TCKEYREC IS ZERO TO PREVENT ERRORS.   */
 /*---------------------------------------------------------------*/
-    apiConnectPtrP->tckeyrec = 0;
+    regApiPtr.p->tckeyrec = 0;
     return;
   default:
     TCKEY_abort(signal, 46);
@@ -4219,34 +4230,46 @@ void Dbtc::diverify010Lab(Signal* signal)
     jam();
     systemErrorLab(signal, __LINE__);
   }//if
-  if (TfirstfreeApiConnectCopy != RNIL) {
-    seizeApiConnectCopy(signal);
-    regApiPtr->apiConnectstate = CS_PREPARE_TO_COMMIT;
-    /*-----------------------------------------------------------------------
-     * WE COME HERE ONLY IF THE TRANSACTION IS PREPARED ON ALL TC CONNECTIONS.
-     * THUS WE CAN START THE COMMIT PHASE BY SENDING DIVERIFY ON ALL TC     
-     * CONNECTIONS AND THEN WHEN ALL DIVERIFYCONF HAVE BEEN RECEIVED THE 
-     * COMMIT MESSAGE CAN BE SENT TO ALL INVOLVED PARTS.
-     *-----------------------------------------------------------------------*/
-    EXECUTE_DIRECT(DBDIH, GSN_DIVERIFYREQ, signal, 1);
-    if (signal->theData[2] == 0) {
-      execDIVERIFYCONF(signal);
+
+  if (regApiPtr->lqhkeyreqrec)
+  {
+    if (TfirstfreeApiConnectCopy != RNIL) {
+      seizeApiConnectCopy(signal);
+      regApiPtr->apiConnectstate = CS_PREPARE_TO_COMMIT;
+      /*-----------------------------------------------------------------------
+       * WE COME HERE ONLY IF THE TRANSACTION IS PREPARED ON ALL TC CONNECTIONS
+       * THUS WE CAN START THE COMMIT PHASE BY SENDING DIVERIFY ON ALL TC     
+       * CONNECTIONS AND THEN WHEN ALL DIVERIFYCONF HAVE BEEN RECEIVED THE 
+       * COMMIT MESSAGE CAN BE SENT TO ALL INVOLVED PARTS.
+       *---------------------------------------------------------------------*/
+      EXECUTE_DIRECT(DBDIH, GSN_DIVERIFYREQ, signal, 1);
+      if (signal->theData[2] == 0) {
+        execDIVERIFYCONF(signal);
+      }
+      return;
+    } else {
+      /*-----------------------------------------------------------------------
+       * There were no free copy connections available. We must abort the 
+       * transaction since otherwise we will have a problem with the report 
+       * to the application.
+       * This should more or less not happen but if it happens we do 
+       * not want to crash and we do not want to create code to handle it 
+       * properly since it is difficult to test it and will be complex to 
+       * handle a problem more or less not occurring.
+       *---------------------------------------------------------------------*/
+      terrorCode = ZSEIZE_API_COPY_ERROR;
+      abortErrorLab(signal);
+      return;
     }
-    return;
-  } else {
-    /*-----------------------------------------------------------------------
-     * There were no free copy connections available. We must abort the 
-     * transaction since otherwise we will have a problem with the report 
-     * to the application.
-     * This should more or less not happen but if it happens we do not want to
-     * crash and we do not want to create code to handle it properly since 
-     * it is difficult to test it and will be complex to handle a problem 
-     * more or less not occurring.
-     *-----------------------------------------------------------------------*/
-    terrorCode = ZSEIZE_API_COPY_ERROR;
-    abortErrorLab(signal);
-    return;
-  }//if
+  }
+  else
+  {
+    jam();
+    sendtckeyconf(signal, 1);
+    regApiPtr->apiConnectstate = CS_CONNECTED;
+    regApiPtr->m_transaction_nodes.clear();
+    setApiConTimer(apiConnectptr.i, 0,__LINE__);
+  }
 }//Dbtc::diverify010Lab()
 
 /* ------------------------------------------------------------------------- */
@@ -5261,16 +5284,8 @@ void Dbtc::execLQHKEYREF(Signal* signal)
       regApiPtr->lqhkeyreqrec--;
       if (regApiPtr->lqhkeyconfrec == regApiPtr->lqhkeyreqrec) {
 	if (regApiPtr->apiConnectstate == CS_START_COMMITTING) {
-	  if(regApiPtr->lqhkeyconfrec) {
-	    jam();
-	    diverify010Lab(signal);
-	  } else {
-	    jam();
-	    sendtckeyconf(signal, 1);
-	    regApiPtr->apiConnectstate = CS_CONNECTED;
-	    regApiPtr->m_transaction_nodes.clear();
-	    setApiConTimer(apiConnectptr.i, 0,__LINE__);
-	  }
+          jam();
+          diverify010Lab(signal);
 	  return;
 	} else if (regApiPtr->tckeyrec > 0 || regApiPtr->m_exec_flag) {
 	  jam();
diff --git a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index 3db91c55849..45d124b8d7d 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
@@ -518,6 +518,7 @@ typedef Ptr<Fragoperrec> FragoperrecPtr;
       Uint32 m_savePointId;
       Uint32 m_scanGCI;
     };
+    Uint32 m_endPage;
     // lock waited for or obtained and not yet passed to LQH
     Uint32 m_accLockOp;
 
@@ -1576,6 +1577,8 @@ public:
 
   void nr_delete_page_callback(Signal*, Uint32 op, Uint32 page);
   void nr_delete_log_buffer_callback(Signal*, Uint32 op, Uint32 page);
+
+  bool get_frag_info(Uint32 tableId, Uint32 fragId, Uint32* maxPage);
 private:
   BLOCK_DEFINES(Dbtup);
 
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
index c8df5f5154e..176efac8058 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
@@ -1464,3 +1464,22 @@ Dbtup::complete_restore_lcp(Uint32 tableId, Uint32 fragId)
     tabDesc += 2;
   }
 }
+
+bool
+Dbtup::get_frag_info(Uint32 tableId, Uint32 fragId, Uint32* maxPage)
+{
+  jamEntry();
+  TablerecPtr tabPtr;
+  tabPtr.i= tableId;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+
+  FragrecordPtr fragPtr;
+  getFragmentrec(fragPtr, fragId, tabPtr.p);
+  
+  if (maxPage)
+  {
+    * maxPage = fragPtr.p->noOfPages;
+  }
+
+  return true;
+}
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
index a5f7d4be0a9..5e9306909b4 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
@@ -95,7 +95,23 @@ Dbtup::execACC_SCANREQ(Signal* signal)
       }
     }
 
-    bits |= AccScanReq::getNRScanFlag(req->requestInfo) ? ScanOp::SCAN_NR : 0;
+    if (AccScanReq::getNRScanFlag(req->requestInfo))
+    {
+      jam();
+      bits |= ScanOp::SCAN_NR;
+      scanPtr.p->m_endPage = req->maxPage;
+      if (req->maxPage != RNIL && req->maxPage > frag.noOfPages)
+      {
+         ndbout_c("%u %u endPage: %u (noOfPages: %u)", 
+                   tablePtr.i, fragId,
+                   req->maxPage, fragPtr.p->noOfPages);
+      }
+    }
+    else
+    {
+      jam();
+      scanPtr.p->m_endPage = RNIL;
+    }
     
     // set up scan op
     new (scanPtr.p) ScanOp();
@@ -540,7 +556,7 @@ Dbtup::scanFirst(Signal*, ScanOpPtr scanPtr)
   ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
   Fragrecord& frag = *fragPtr.p;
   // in the future should not pre-allocate pages
-  if (frag.noOfPages == 0) {
+  if (frag.noOfPages == 0 && ((bits & ScanOp::SCAN_NR) == 0)) {
     jam();
     scan.m_state = ScanOp::Last;
     return;
@@ -632,11 +648,23 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
         key.m_page_no++;
         if (key.m_page_no >= frag.noOfPages) {
           jam();
+
+          if ((bits & ScanOp::SCAN_NR) && (scan.m_endPage != RNIL))
+          {
+            jam();
+            if (key.m_page_no < scan.m_endPage)
+            {
+              jam();
+              ndbout_c("scanning page %u", key.m_page_no);
+              goto cont;
+            }
+          }
           // no more pages, scan ends
           pos.m_get = ScanPos::Get_undef;
           scan.m_state = ScanOp::Last;
           return true;
         }
+    cont:
         key.m_page_idx = 0;
         pos.m_get = ScanPos::Get_page_mm;
         // clear cached value
@@ -649,7 +677,13 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
       {
         if (pos.m_realpid_mm == RNIL) {
           jam();
-          pos.m_realpid_mm = getRealpid(fragPtr.p, key.m_page_no);
+          if (key.m_page_no < frag.noOfPages)
+            pos.m_realpid_mm = getRealpid(fragPtr.p, key.m_page_no);
+          else
+          {
+            ndbassert(bits & ScanOp::SCAN_NR);
+            goto nopage;
+          }
         }
         PagePtr pagePtr;
 	c_page_pool.getPtr(pagePtr, pos.m_realpid_mm);
@@ -657,9 +691,18 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
         if (pagePtr.p->page_state == ZEMPTY_MM) {
           // skip empty page
           jam();
-          pos.m_get = ScanPos::Get_next_page_mm;
-          break; // incr loop count
+          if (! (bits & ScanOp::SCAN_NR))
+          {
+            pos.m_get = ScanPos::Get_next_page_mm;
+            break; // incr loop count
+          }
+          else
+          {
+            jam();
+            pos.m_realpid_mm = RNIL;
+          }
         }
+    nopage:
         pos.m_page = pagePtr.p;
         pos.m_get = ScanPos::Get_tuple;
       }
@@ -820,11 +863,11 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
 	{
 	  pos.m_get = ScanPos::Get_next_tuple_fs;
           th = (Tuple_header*)&page->m_data[key.m_page_idx];
-	  thbits = th->m_header_bits;
 	  
 	  if (likely(! (bits & ScanOp::SCAN_NR)))
 	  {
 	    jam();
+            thbits = th->m_header_bits;
 	    if (! (thbits & Tuple_header::FREE))
 	    {
               goto found_tuple;
@@ -832,7 +875,15 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
 	  }
 	  else
 	  {
-	    if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI)
+            if (pos.m_realpid_mm == RNIL)
+            {
+              jam();
+              foundGCI = 0;
+              goto found_deleted_rowid;
+            }
+            thbits = th->m_header_bits;
+	    if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI ||
+                foundGCI == 0)
 	    {
 	      if (! (thbits & Tuple_header::FREE))
 	      {
@@ -904,7 +955,8 @@ Dbtup::scanNext(Signal* signal, ScanOpPtr scanPtr)
 	  
 	  Fix_page *mmpage = (Fix_page*)c_page_pool.getPtr(pos.m_realpid_mm);
 	  th = (Tuple_header*)(mmpage->m_data + key_mm.m_page_idx);
-	  if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI)
+	  if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI ||
+              foundGCI == 0)
 	  {
 	    if (! (thbits & Tuple_header::FREE))
 	      break;
diff --git a/storage/ndb/src/kernel/blocks/suma/Suma.cpp b/storage/ndb/src/kernel/blocks/suma/Suma.cpp
index b8cc438f514..7845b83693c 100644
--- a/storage/ndb/src/kernel/blocks/suma/Suma.cpp
+++ b/storage/ndb/src/kernel/blocks/suma/Suma.cpp
@@ -3655,6 +3655,8 @@ Suma::execSUB_GCP_COMPLETE_REP(Signal* signal)
   
   if(m_gcp_complete_rep_count && !c_subscriber_nodes.isclear())
   {
+    CRASH_INSERTION(13033);
+    
     NodeReceiverGroup rg(API_CLUSTERMGR, c_subscriber_nodes);
     sendSignal(rg, GSN_SUB_GCP_COMPLETE_REP, signal,
 	       SubGcpCompleteRep::SignalLength, JBB);
@@ -3674,8 +3676,8 @@ Suma::execSUB_GCP_COMPLETE_REP(Signal* signal)
   {
     if(m_active_buckets.get(i))
       continue;
-    
-    if(c_buckets[i].m_buffer_tail != RNIL)
+
+    if (!c_subscriber_nodes.isclear())
     {
       //Uint32* dst;
       get_buffer_ptr(signal, i, gci, 0);
diff --git a/storage/ndb/src/ndbapi/NdbBlob.cpp b/storage/ndb/src/ndbapi/NdbBlob.cpp
index 0fc96add084..f3d1dbe3dd1 100644
--- a/storage/ndb/src/ndbapi/NdbBlob.cpp
+++ b/storage/ndb/src/ndbapi/NdbBlob.cpp
@@ -1141,7 +1141,7 @@ NdbBlob::readTableParts(char* buf, Uint32 part, Uint32 count)
          * table tuple does not fully protect blob parts since DBTUP
          * commits each tuple separately.
          */
-        tOp->readTuple() == -1 ||
+        tOp->readTuple(NdbOperation::LM_SimpleRead) == -1 ||
         setPartKeyValue(tOp, part + n) == -1 ||
         tOp->getValue((Uint32)3, buf) == NULL) {
       setErrorCode(tOp);
diff --git a/storage/ndb/src/ndbapi/NdbIndexOperation.cpp b/storage/ndb/src/ndbapi/NdbIndexOperation.cpp
index fc19bd251d4..921769f09e3 100644
--- a/storage/ndb/src/ndbapi/NdbIndexOperation.cpp
+++ b/storage/ndb/src/ndbapi/NdbIndexOperation.cpp
@@ -85,6 +85,9 @@ int NdbIndexOperation::readTuple(NdbOperation::LockMode lm)
   case LM_CommittedRead:
     return readTuple();
     break;
+  case LM_SimpleRead:
+    return readTuple();
+    break;
   default:
     return -1;
   };
diff --git a/storage/ndb/src/ndbapi/NdbOperationDefine.cpp b/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
index c9459ff911c..73778ab9bbe 100644
--- a/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
+++ b/storage/ndb/src/ndbapi/NdbOperationDefine.cpp
@@ -131,6 +131,8 @@ NdbOperation::readTuple(NdbOperation::LockMode lm)
   case LM_CommittedRead:
     return committedRead();
     break;
+  case LM_SimpleRead:
+    return simpleRead();
   default:
     return -1;
   };
@@ -185,24 +187,22 @@ NdbOperation::readTupleExclusive()
 int
 NdbOperation::simpleRead()
 {
-  /**
-   * Currently/still disabled
-   */
-  return readTuple();
-#if 0
+  NdbTransaction* tNdbCon = theNdbCon;
   int tErrorLine = theErrorLine;
   if (theStatus == Init) {
     theStatus = OperationDefined;
     theOperationType = ReadRequest;
     theSimpleIndicator = 1;
+    theDirtyIndicator = 0;
     theErrorLine = tErrorLine++;
-    theLockMode = LM_Read;
+    theLockMode = LM_SimpleRead;
+    m_abortOption = AO_IgnoreError;
+    tNdbCon->theSimpleState = 0;
     return 0;
   } else {
     setErrorCode(4200);
     return -1;
   }//if
-#endif
 }//NdbOperation::simpleRead()
 
 /*****************************************************************************
@@ -338,28 +338,32 @@ 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);
+  switch (lockMode) {
+  case LM_CommittedRead: /* TODO, check theNdbCon->theSimpleState */
+    theOperationType= ReadRequest;
+    theSimpleIndicator= 1;
+    theDirtyIndicator= 1;
+    break;
+  case LM_SimpleRead: /* TODO, check theNdbCon->theSimpleState */
+    theOperationType= ReadRequest;
+    theSimpleIndicator= 1;
+    theDirtyIndicator= 0;
+    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/NdbOperationExec.cpp b/storage/ndb/src/ndbapi/NdbOperationExec.cpp
index 9fe85265a0c..27672e0458c 100644
--- a/storage/ndb/src/ndbapi/NdbOperationExec.cpp
+++ b/storage/ndb/src/ndbapi/NdbOperationExec.cpp
@@ -175,12 +175,11 @@ NdbOperation::prepareSend(Uint32 aTC_ConnectPtr,
   Uint8 tInterpretIndicator = theInterpretIndicator;
   Uint8 tNoDisk = m_no_disk_flag;
 
-//-------------------------------------------------------------
-// Simple state is set if start and commit is set and it is
-// a read request. Otherwise it is set to zero.
-//-------------------------------------------------------------
+  /**
+   * A dirty read, can not abort the transaction
+   */
   Uint8 tReadInd = (theOperationType == ReadRequest);
-  Uint8 tSimpleState = tReadInd & tSimpleIndicator;
+  Uint8 tDirtyState = tReadInd & tDirtyIndicator;
 
   tcKeyReq->transId1           = tTransId1;
   tcKeyReq->transId2           = tTransId2;
@@ -206,8 +205,8 @@ NdbOperation::prepareSend(Uint32 aTC_ConnectPtr,
   tcKeyReq->setOperationType(tReqInfo, tOperationType);
   tcKeyReq->setKeyLength(tReqInfo, tTupKeyLen);
   
-  // A simple read is always ignore error
-  abortOption = tSimpleState ? (Uint8) AO_IgnoreError : (Uint8) abortOption;
+  // A dirty read is always ignore error
+  abortOption = tDirtyState ? (Uint8) AO_IgnoreError : (Uint8) abortOption;
   tcKeyReq->setAbortOption(tReqInfo, abortOption);
   m_abortOption = abortOption;
   
@@ -549,8 +548,8 @@ NdbOperation::receiveTCKEYREF( NdbApiSignal* aSignal)
   theStatus = Finished;
   theReceiver.m_received_result_length = ~0;
 
-  // not simple read
-  if(! (theOperationType == ReadRequest && theSimpleIndicator))
+  // not dirty read
+  if(! (theOperationType == ReadRequest && theDirtyIndicator))
   {
     theNdbCon->OpCompleteFailure(this);
     return -1;
diff --git a/storage/ndb/src/ndbapi/NdbReceiver.cpp b/storage/ndb/src/ndbapi/NdbReceiver.cpp
index 34b3a14ac6e..5a311bcbefe 100644
--- a/storage/ndb/src/ndbapi/NdbReceiver.cpp
+++ b/storage/ndb/src/ndbapi/NdbReceiver.cpp
@@ -283,7 +283,7 @@ NdbReceiver::execTRANSID_AI(const Uint32* aDataPtr, Uint32 aLength)
   Uint32 tmp = m_received_result_length + aLength;
   m_received_result_length = tmp;
 
-  return (tmp == exp || (exp > TcKeyConf::SimpleReadBit) ? 1 : 0);
+  return (tmp == exp || (exp > TcKeyConf::DirtyReadBit) ? 1 : 0);
 }
 
 int
diff --git a/storage/ndb/src/ndbapi/NdbScanOperation.cpp b/storage/ndb/src/ndbapi/NdbScanOperation.cpp
index 055fbf28339..afbec070ac8 100644
--- a/storage/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/storage/ndb/src/ndbapi/NdbScanOperation.cpp
@@ -237,6 +237,7 @@ NdbScanOperation::setReadLockMode(LockMode lockMode)
       lockHoldMode= false;
       readCommitted= true;
       break;
+    case LM_SimpleRead:
     case LM_Read:
       lockExcl= false;
       lockHoldMode= true;
diff --git a/storage/ndb/src/ndbapi/NdbTransaction.cpp b/storage/ndb/src/ndbapi/NdbTransaction.cpp
index 55c6f0f4b99..bc59df722aa 100644
--- a/storage/ndb/src/ndbapi/NdbTransaction.cpp
+++ b/storage/ndb/src/ndbapi/NdbTransaction.cpp
@@ -1787,8 +1787,8 @@ from other transactions.
       const Uint32 tAttrInfoLen = *tPtr++;
       if (tOp && tOp->checkMagicNumber()) {
 	Uint32 done = tOp->execTCOPCONF(tAttrInfoLen);
-	if(tAttrInfoLen > TcKeyConf::SimpleReadBit){
-	  Uint32 node = tAttrInfoLen & (~TcKeyConf::SimpleReadBit);
+	if(tAttrInfoLen > TcKeyConf::DirtyReadBit){
+	  Uint32 node = tAttrInfoLen & (~TcKeyConf::DirtyReadBit);
 	  NdbNodeBitmask::set(m_db_nodes, node);
 	  if(NdbNodeBitmask::get(m_failed_db_nodes, node) && !done)
 	  {
@@ -2182,7 +2182,7 @@ NdbTransaction::report_node_failure(Uint32 id){
    * 4)   X           X
    */
   NdbOperation* tmp = theFirstExecOpInList;
-  const Uint32 len = TcKeyConf::SimpleReadBit | id;
+  const Uint32 len = TcKeyConf::DirtyReadBit | id;
   Uint32 tNoComp = theNoOfOpCompleted;
   Uint32 tNoSent = theNoOfOpSent;
   Uint32 count = 0;
diff --git a/storage/ndb/test/ndbapi/testBasic.cpp b/storage/ndb/test/ndbapi/testBasic.cpp
index 952b5a50dc5..ac23ceaad18 100644
--- a/storage/ndb/test/ndbapi/testBasic.cpp
+++ b/storage/ndb/test/ndbapi/testBasic.cpp
@@ -136,31 +136,13 @@ int runPkRead(NDBT_Context* ctx, NDBT_Step* step){
   int loops = ctx->getNumLoops();
   int records = ctx->getNumRecords();
   int batchSize = ctx->getProperty("BatchSize", 1);
+  int lm = ctx->getProperty("LockMode", NdbOperation::LM_Read);
   int i = 0;
   HugoTransactions hugoTrans(*ctx->getTab());
   while (i<loops) {
     g_info << i << ": ";
-    if (hugoTrans.pkReadRecords(GETNDB(step), records, batchSize) != NDBT_OK){
-      g_info << endl;
-      return NDBT_FAILED;
-    }
-    i++;
-  }
-  g_info << endl;
-  return NDBT_OK;
-}
-
-int runPkDirtyRead(NDBT_Context* ctx, NDBT_Step* step){
-  int loops = ctx->getNumLoops();
-  int records = ctx->getNumRecords();
-  int batchSize = ctx->getProperty("BatchSize", 1);
-  int i = 0;
-  bool dirty = true;
-  HugoTransactions hugoTrans(*ctx->getTab());
-  while (i<loops) {
-    g_info << i << ": ";
-    if (hugoTrans.pkReadRecords(GETNDB(step), records, batchSize, 
-				NdbOperation::LM_CommittedRead) != NDBT_OK){
+    if (hugoTrans.pkReadRecords(GETNDB(step), records, batchSize,
+                                (NdbOperation::LockMode)lm) != NDBT_OK){
       g_info << endl;
       return NDBT_FAILED;
     }
@@ -1552,14 +1534,23 @@ TESTCASE("PkInsert",
 }
 TESTCASE("PkRead", 
 	   "Verify that we can insert, read and delete from this table using PK"){
+  TC_PROPERTY("LockMode", NdbOperation::LM_Read);
   INITIALIZER(runLoadTable);
   STEP(runPkRead);
   FINALIZER(runClearTable);
 }
 TESTCASE("PkDirtyRead", 
 	 "Verify that we can insert, dirty read and delete from this table using PK"){
+  TC_PROPERTY("LockMode", NdbOperation::LM_Dirty);
   INITIALIZER(runLoadTable);
-  STEP(runPkDirtyRead);
+  STEP(runPkRead);
+  FINALIZER(runClearTable);
+}
+TESTCASE("PkSimpleRead", 
+	 "Verify that we can insert, simple read and delete from this table using PK"){
+  TC_PROPERTY("LockMode", NdbOperation::LM_SimpleRead);
+  INITIALIZER(runLoadTable);
+  STEP(runPkRead);
   FINALIZER(runClearTable);
 }
 TESTCASE("PkUpdate", 
diff --git a/storage/ndb/test/ndbapi/testSystemRestart.cpp b/storage/ndb/test/ndbapi/testSystemRestart.cpp
index 3cd7a3798c5..89580c0cef8 100644
--- a/storage/ndb/test/ndbapi/testSystemRestart.cpp
+++ b/storage/ndb/test/ndbapi/testSystemRestart.cpp
@@ -1501,6 +1501,54 @@ int runSR_DD_2(NDBT_Context* ctx, NDBT_Step* step)
   return result;
 }
 
+int 
+runBug27434(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int result = NDBT_OK;
+  NdbRestarter restarter;
+  Ndb* pNdb = GETNDB(step);
+  const Uint32 nodeCount = restarter.getNumDbNodes();
+
+  if (nodeCount < 2)
+    return NDBT_OK;
+
+  int args[] = { DumpStateOrd::DihMaxTimeBetweenLCP };
+  int dump[] = { DumpStateOrd::DihStartLcpImmediately };
+
+  int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_CHECKPOINT, 0 };
+  NdbLogEventHandle handle = 
+    ndb_mgm_create_logevent_handle(restarter.handle, filter);
+
+  struct ndb_logevent event;
+
+  do {
+    int node1 = restarter.getDbNodeId(rand() % nodeCount);
+    CHECK(restarter.restartOneDbNode(node1, false, true, true) == 0);
+    NdbSleep_SecSleep(3);
+    CHECK(restarter.waitNodesNoStart(&node1, 1) == 0);
+
+    CHECK(restarter.dumpStateAllNodes(args, 1) == 0);
+
+    for (Uint32 i = 0; i<3; i++)
+    {
+      CHECK(restarter.dumpStateAllNodes(dump, 1) == 0);
+      while(ndb_logevent_get_next(handle, &event, 0) >= 0 &&
+	    event.type != NDB_LE_LocalCheckpointStarted);
+      while(ndb_logevent_get_next(handle, &event, 0) >= 0 &&
+	    event.type != NDB_LE_LocalCheckpointCompleted);
+    }      
+    
+    restarter.restartAll(false, true, true);
+    NdbSleep_SecSleep(3);
+    CHECK(restarter.waitClusterNoStart() == 0);
+    restarter.insertErrorInNode(node1, 5046);
+    restarter.startAll();
+    CHECK(restarter.waitClusterStarted() == 0);
+  } while(false);
+  
+  return result;
+}
+
 NDBT_TESTSUITE(testSystemRestart);
 TESTCASE("SR1", 
 	 "Basic system restart test. Focus on testing restart from REDO log.\n"
@@ -1681,6 +1729,12 @@ TESTCASE("Bug24664",
   STEP(runBug24664);
   FINALIZER(runClearTable);
 }
+TESTCASE("Bug27434",
+	 "")
+{
+  INITIALIZER(runWaitStarted);
+  STEP(runBug27434);
+}
 TESTCASE("SR_DD_1", "")
 {
   TC_PROPERTY("ALL", 1);
diff --git a/storage/ndb/test/ndbapi/test_event.cpp b/storage/ndb/test/ndbapi/test_event.cpp
index 2083e235a3e..18825d734a4 100644
--- a/storage/ndb/test/ndbapi/test_event.cpp
+++ b/storage/ndb/test/ndbapi/test_event.cpp
@@ -21,6 +21,7 @@
 #include <NdbAutoPtr.hpp>
 #include <NdbRestarter.hpp>
 #include <NdbRestarts.hpp>
+#include <signaldata/DumpStateOrd.hpp>
 
 #define GETNDB(ps) ((NDBT_NdbApiStep*)ps)->getNdb()
 
@@ -1758,6 +1759,85 @@ runInsertDeleteUntilStopped(NDBT_Context* ctx, NDBT_Step* step)
   return NDBT_OK;
 }
 
+int 
+runBug31701(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int result = NDBT_OK;
+
+  NdbRestarter restarter;
+
+  if (restarter.getNumDbNodes() < 2){
+    ctx->stopTest();
+    return NDBT_OK;
+  }
+  // This should really wait for applier to start...10s is likely enough
+  NdbSleep_SecSleep(10);
+
+  int nodeId = restarter.getDbNodeId(rand() % restarter.getNumDbNodes());
+
+  int val2[] = { DumpStateOrd::CmvmiSetRestartOnErrorInsert, 1 };
+  if (restarter.dumpStateOneNode(nodeId, val2, 2))
+    return NDBT_FAILED;
+  
+  restarter.insertErrorInNode(nodeId, 13033);
+  if (restarter.waitNodesNoStart(&nodeId, 1))
+    return NDBT_FAILED;
+
+  if (restarter.startNodes(&nodeId, 1))
+    return NDBT_FAILED;
+
+  if (restarter.waitClusterStarted())
+    return NDBT_FAILED;
+
+  
+  int records = ctx->getNumRecords();
+  HugoTransactions hugoTrans(*ctx->getTab());
+  
+  if(ctx->getPropertyWait("LastGCI", ~(Uint32)0))
+  {
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+
+  hugoTrans.clearTable(GETNDB(step), 0);
+  
+  if (hugoTrans.loadTable(GETNDB(step), 3*records, 1, true, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  
+  if (hugoTrans.pkDelRecords(GETNDB(step), 3*records, 1, true, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  if (hugoTrans.loadTable(GETNDB(step), records, 1, true, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  if (hugoTrans.pkUpdateRecords(GETNDB(step), records, 1, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  if (hugoTrans.pkUpdateRecords(GETNDB(step), records, 1, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  if (hugoTrans.pkUpdateRecords(GETNDB(step), records, 1, 1) != 0){
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+  
+  ctx->setProperty("LastGCI", hugoTrans.m_latest_gci);
+  if(ctx->getPropertyWait("LastGCI", ~(Uint32)0))
+  {
+    g_err << "FAIL " << __LINE__ << endl;
+    return NDBT_FAILED;
+  }
+
+  ctx->stopTest();  
+  return NDBT_OK;
+}
+
 NDBT_TESTSUITE(test_event);
 TESTCASE("BasicEventOperation", 
 	 "Verify that we can listen to Events"
@@ -1887,6 +1967,14 @@ TESTCASE("Bug27169", ""){
   STEP(runRestarterLoop);
   FINALIZER(runDropEvent);
 }
+TESTCASE("Bug31701", ""){
+  INITIALIZER(runCreateEvent);
+  INITIALIZER(runCreateShadowTable);
+  STEP(runEventApplier);
+  STEP(runBug31701);
+  FINALIZER(runDropEvent);
+  FINALIZER(runDropShadowTable);
+}
 NDBT_TESTSUITE_END(test_event);
 
 int main(int argc, const char** argv){
diff --git a/storage/ndb/test/run-test/daily-basic-tests.txt b/storage/ndb/test/run-test/daily-basic-tests.txt
index 0ca81b7f63b..103675d8e35 100644
--- a/storage/ndb/test/run-test/daily-basic-tests.txt
+++ b/storage/ndb/test/run-test/daily-basic-tests.txt
@@ -63,6 +63,14 @@ max-time: 500
 cmd: testBasic
 args: -n PkRead
 
+max-time: 500
+cmd: testBasic
+args: -n PkSimpleRead
+
+max-time: 500
+cmd: testBasic
+args: -n PkDirtyRead
+
 max-time: 500
 cmd: testBasic
 args: -n PkUpdate 
@@ -872,6 +880,10 @@ max-time: 1000
 cmd: testNodeRestart
 args: -n Bug27466 T1
 
+max-time: 1500
+cmd: testSystemRestart
+args: -n Bug27434 T1
+
 max-time: 1000
 cmd: test_event
 args: -l 10 -n Bug27169 T1
@@ -1022,3 +1034,8 @@ max-time: 600
 cmd: testNodeRestart
 args: -n Bug31525 T1
 
+max-time: 300
+cmd: test_event
+args: -n Bug31701 T1
+
+
diff --git a/storage/ndb/test/src/HugoOperations.cpp b/storage/ndb/test/src/HugoOperations.cpp
index 1a2e5180f1f..93a9eaf435a 100644
--- a/storage/ndb/test/src/HugoOperations.cpp
+++ b/storage/ndb/test/src/HugoOperations.cpp
@@ -93,6 +93,7 @@ rand_lock_mode:
     case NdbOperation::LM_Read:
     case NdbOperation::LM_Exclusive:
     case NdbOperation::LM_CommittedRead:
+    case NdbOperation::LM_SimpleRead:
       if(idx && idx->getType() == NdbDictionary::Index::OrderedIndex && 
 	 pIndexScanOp == 0)
       {
diff --git a/storage/ndb/test/src/NdbRestarts.cpp b/storage/ndb/test/src/NdbRestarts.cpp
index 6ec520887b5..86e71f4b3fc 100644
--- a/storage/ndb/test/src/NdbRestarts.cpp
+++ b/storage/ndb/test/src/NdbRestarts.cpp
@@ -607,6 +607,7 @@ NFDuringNR_codes[] = {
   5026,
   7139,
   7132,
+  5045,
 
   //LCP
   8000,