Added new testcase for Bug #4479

testBasic -n MassiveTransaction
Inserts as many records as defined in one transaction
using loadTable

ndb/test/include/HugoTransactions.hpp

@@ -34,7 +34,8 @@ public:
 		int records,
 		int batch = 512,
 		bool allowConstraintViolation = true,
-		int doSleep = 0);
+		int doSleep = 0,
+                bool oneTrans = false);
   int scanReadRecords(Ndb*, 
 		      int records,
 		      int abort = 0,

ndb/test/ndbapi/testBasic.cpp

@@ -29,9 +29,18 @@
  *  delete should be visible to same transaction
  *  
  */
+int runLoadTable2(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int records = ctx->getNumRecords();
+  HugoTransactions hugoTrans(*ctx->getTab());
+  if (hugoTrans.loadTable(GETNDB(step), records, 512, false, 0, true) != 0){
+    return NDBT_FAILED;
+  }
+  return NDBT_OK;
+}
 
-int runLoadTable(NDBT_Context* ctx, NDBT_Step* step){
-
+int runLoadTable(NDBT_Context* ctx, NDBT_Step* step)
+{
   int records = ctx->getNumRecords();
   HugoTransactions hugoTrans(*ctx->getTab());
   if (hugoTrans.loadTable(GETNDB(step), records) != 0){
@@ -1255,6 +1264,11 @@ TESTCASE("MassiveRollback2",
   INITIALIZER(runMassiveRollback2);
   FINALIZER(runClearTable2);
 }
+TESTCASE("MassiveTransaction",
+         "Test very large insert transaction"){
+  INITIALIZER(runLoadTable2);
+  FINALIZER(runClearTable2);
+}
 NDBT_TESTSUITE_END(testBasic);
 
 int main(int argc, const char** argv){

ndb/test/src/HugoTransactions.cpp

@@ -819,12 +819,14 @@ HugoTransactions::loadTable(Ndb* pNdb,
 			    int records,
 			    int batch,
 			    bool allowConstraintViolation,
-			    int doSleep){
+			    int doSleep,
+                            bool oneTrans){
   int             check;
   int             retryAttempt = 0;
   int             retryMax = 5;
   NdbConnection   *pTrans;
   NdbOperation	  *pOp;
+  bool            first_batch = true;
 
   const int org = batch;
   const int cols = tab.getNoOfColumns();
@@ -833,7 +835,7 @@ HugoTransactions::loadTable(Ndb* pNdb,
   batch = (batch * 256); // -> 512 -> 65536k per commit
   batch = batch/bytes;   // 
   batch = batch == 0 ? 1 : batch;
-  
+ 
   if(batch != org){
     g_info << "batch = " << org << " rowsize = " << bytes
 	   << " -> rows/commit = " << batch << endl;
@@ -841,7 +843,7 @@ HugoTransactions::loadTable(Ndb* pNdb,
   
   g_info << "|- Inserting records..." << endl;
   for (int c=0 ; c<records ; ){
-
+    bool closeTrans;
     if (retryAttempt >= retryMax){
       g_info << "Record " << c << " could not be inserted, has retried "
 	     << retryAttempt << " times " << endl;
@@ -852,19 +854,22 @@ HugoTransactions::loadTable(Ndb* pNdb,
     if (doSleep > 0)
       NdbSleep_MilliSleep(doSleep);
 
-    pTrans = pNdb->startTransaction();
-
-    if (pTrans == NULL) {
-      const NdbError err = pNdb->getNdbError();
+    if (first_batch || !oneTrans) {
+      first_batch = false;
+      pTrans = pNdb->startTransaction();
+    
+      if (pTrans == NULL) {
+        const NdbError err = pNdb->getNdbError();
 
-      if (err.status == NdbError::TemporaryError){
-	ERR(err);
-	NdbSleep_MilliSleep(50);
-	retryAttempt++;
-	continue;
+        if (err.status == NdbError::TemporaryError){
+          ERR(err);
+	  NdbSleep_MilliSleep(50);
+	  retryAttempt++;
+	  continue;
+        }
+        ERR(err);
+        return NDBT_FAILED;
       }
-      ERR(err);
-      return NDBT_FAILED;
     }
 
     for(int b = 0; b < batch && c+b<records; b++){ 
@@ -894,7 +899,13 @@ HugoTransactions::loadTable(Ndb* pNdb,
     }
     
     // Execute the transaction and insert the record
-    check = pTrans->execute( Commit ); 
+    if (!oneTrans || (c + batch) >= records) {
+      closeTrans = true;
+      check = pTrans->execute( Commit );
+    } else {
+      closeTrans = false;
+      check = pTrans->execute( NoCommit );
+    }
     if(check == -1 ) {
       const NdbError err = pTrans->getNdbError();
       pNdb->closeTransaction(pTrans);
@@ -937,8 +948,10 @@ HugoTransactions::loadTable(Ndb* pNdb,
 	break;
       }
     }
-    else{      
-      pNdb->closeTransaction(pTrans);
+    else{
+      if (closeTrans) {
+        pNdb->closeTransaction(pTrans);
+      }
     }
     
     // Step to next record