ndb - bug#20847 : DbtupTabDesMan: add merge with left buddies

ndb/src/kernel/blocks/ERROR_codes.txt:
  DbtupTabDesMan: add merge with left buddies
ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  DbtupTabDesMan: add merge with left buddies
ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp:
  DbtupTabDesMan: add merge with left buddies
ndb/src/kernel/blocks/dbtup/DbtupTabDesMan.cpp:
  DbtupTabDesMan: add merge with left buddies
ndb/test/ndbapi/testDict.cpp:
  DbtupTabDesMan: add merge with left buddies
ndb/test/run-test/daily-basic-tests.txt:
  DbtupTabDesMan: add merge with left buddies

6 files changed, 257 insertions, 34 deletions

diff --git a/ndb/src/kernel/blocks/ERROR_codes.txt b/ndb/src/kernel/blocks/ERROR_codes.txt
index 20f03e7ea69..de32ce91ee8 100644
--- a/ndb/src/kernel/blocks/ERROR_codes.txt
+++ b/ndb/src/kernel/blocks/ERROR_codes.txt
@@ -2,7 +2,7 @@ Next QMGR 1
 Next NDBCNTR 1000
 Next NDBFS 2000
 Next DBACC 3002
-Next DBTUP 4013
+Next DBTUP 4014
 Next DBLQH 5043
 Next DBDICT 6006
 Next DBDIH 7174
@@ -430,6 +430,8 @@ Drop Table/Index:
 8035: Fail next trigger drop in TC
 8036: Fail next index drop in TC
 
+4013: verify TUP tab descr before and after next DROP TABLE
+
 System Restart:
 ---------------
 
diff --git a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index 1cb3bd89997..360710d543b 100644
--- a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
@@ -2129,15 +2129,18 @@ private:
 // Public methods
   Uint32 getTabDescrOffsets(const Tablerec* regTabPtr, Uint32* offset);
   Uint32 allocTabDescr(const Tablerec* regTabPtr, Uint32* offset);
-  void freeTabDescr(Uint32 retRef, Uint32 retNo);
+  void freeTabDescr(Uint32 retRef, Uint32 retNo, bool normal = true);
   Uint32 getTabDescrWord(Uint32 index);
   void setTabDescrWord(Uint32 index, Uint32 word);
 
 // Private methods
   Uint32 sizeOfReadFunction();
   void   removeTdArea(Uint32 tabDesRef, Uint32 list);
-  void   insertTdArea(Uint32 sizeOfChunk, Uint32 tabDesRef, Uint32 list);
-  Uint32 itdaMergeTabDescr(Uint32 retRef, Uint32 retNo);
+  void   insertTdArea(Uint32 tabDesRef, Uint32 list);
+  void   itdaMergeTabDescr(Uint32& retRef, Uint32& retNo, bool normal);
+#ifdef VM_TRACE
+  void verifytabdes();
+#endif
 
 //------------------------------------------------------------------------------------------------------
 // Page Memory Manager
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
index c6e33bdc92b..ad8daff0729 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
+++ b/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
@@ -567,6 +567,9 @@ void
 Dbtup::execDROP_TAB_REQ(Signal* signal)
 {
   ljamEntry();
+  if (ERROR_INSERTED(4013)) {
+    verifytabdes();
+  }
   DropTabReq* req = (DropTabReq*)signal->getDataPtr();
   
   TablerecPtr tabPtr;
@@ -685,5 +688,9 @@ void Dbtup::execFSREMOVECONF(Signal* signal)
   
   releaseTabDescr(tabPtr.p);
   initTab(tabPtr.p);
+  if (ERROR_INSERTED(4013)) {
+    CLEAR_ERROR_INSERT_VALUE;
+    verifytabdes();
+  }
 }//Dbtup::execFSREMOVECONF()
 
diff --git a/ndb/src/kernel/blocks/dbtup/DbtupTabDesMan.cpp b/ndb/src/kernel/blocks/dbtup/DbtupTabDesMan.cpp
index e74ac93f3f8..3e96bc6c14a 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupTabDesMan.cpp
+++ b/ndb/src/kernel/blocks/dbtup/DbtupTabDesMan.cpp
@@ -24,13 +24,15 @@
 #define ljam() { jamLine(22000 + __LINE__); }
 #define ljamEntry() { jamEntryLine(22000 + __LINE__); }
 
-/* **************************************************************** */
-/* *********** TABLE DESCRIPTOR MEMORY MANAGER ******************** */
-/* **************************************************************** */
-/* This module is used to allocate and deallocate table descriptor  */
-/* memory attached to fragments (could be allocated per table       */
-/* instead. Performs its task by a buddy algorithm.                 */
-/* **************************************************************** */
+/*
+ * TABLE DESCRIPTOR MEMORY MANAGER
+ *
+ * Each table has a descriptor which is a contiguous array of words.
+ * The descriptor is allocated from a global array using a buddy
+ * algorithm.  Free lists exist for each power of 2 words.  Freeing
+ * a piece first merges with free right and left neighbours and then
+ * divides itself up into free list chunks.
+ */
 
 Uint32
 Dbtup::getTabDescrOffsets(const Tablerec* regTabPtr, Uint32* offset)
@@ -72,8 +74,9 @@ Uint32 Dbtup::allocTabDescr(const Tablerec* regTabPtr, Uint32* offset)
       Uint32 retNo = (1 << i) - allocSize;	        /* CALCULATE THE DIFFERENCE               */
       if (retNo >= ZTD_FREE_SIZE) {
         ljam();
-        Uint32 retRef = reference + allocSize;          /* SET THE RETURN POINTER                 */
-        freeTabDescr(retRef, retNo);	                /* RETURN UNUSED TD SPACE TO THE TD AREA  */
+        // return unused words, of course without attempting left merge
+        Uint32 retRef = reference + allocSize;
+        freeTabDescr(retRef, retNo, false);
       } else {
         ljam();
         allocSize = 1 << i;
@@ -99,15 +102,15 @@ Uint32 Dbtup::allocTabDescr(const Tablerec* regTabPtr, Uint32* offset)
   }//if
 }//Dbtup::allocTabDescr()
 
-void Dbtup::freeTabDescr(Uint32 retRef, Uint32 retNo) 
+void Dbtup::freeTabDescr(Uint32 retRef, Uint32 retNo, bool normal)
 {
-  retNo = itdaMergeTabDescr(retRef, retNo);       /* MERGE WITH POSSIBLE RIGHT NEIGHBOURS   */
+  itdaMergeTabDescr(retRef, retNo, normal);       /* MERGE WITH POSSIBLE NEIGHBOURS   */
   while (retNo >= ZTD_FREE_SIZE) {
     ljam();
     Uint32 list = nextHigherTwoLog(retNo);
     list--;	/* RETURN TO NEXT LOWER LIST    */
     Uint32 sizeOfChunk = 1 << list;
-    insertTdArea(sizeOfChunk, retRef, list);
+    insertTdArea(retRef, list);
     retRef += sizeOfChunk;
     retNo -= sizeOfChunk;
   }//while
@@ -128,7 +131,7 @@ Dbtup::setTabDescrWord(Uint32 index, Uint32 word)
   tableDescriptor[index].tabDescr = word;
 }//Dbtup::setTabDescrWord()
 
-void Dbtup::insertTdArea(Uint32 sizeOfChunk, Uint32 tabDesRef, Uint32 list) 
+void Dbtup::insertTdArea(Uint32 tabDesRef, Uint32 list) 
 {
   ndbrequire(list < 16);
   setTabDescrWord(tabDesRef + ZTD_FL_HEADER, ZTD_TYPE_FREE);
@@ -145,19 +148,14 @@ void Dbtup::insertTdArea(Uint32 sizeOfChunk, Uint32 tabDesRef, Uint32 list)
   setTabDescrWord((tabDesRef + (1 << list)) - ZTD_TR_SIZE, 1 << list);
 }//Dbtup::insertTdArea()
 
-/* ---------------------------------------------------------------- */
-/* ----------------------- MERGE_TAB_DESCR ------------------------ */
-/* ---------------------------------------------------------------- */
-/* INPUT:  TAB_DESCR_PTR   POINTING AT THE CURRENT CHUNK            */
-/*                                                                  */
-/* SHORTNAME:   MTD                                                 */
-/* -----------------------------------------------------------------*/
-Uint32 Dbtup::itdaMergeTabDescr(Uint32 retRef, Uint32 retNo) 
+/*
+ * Merge to-be-removed chunk (which need not be initialized with header
+ * and trailer) with left and right buddies.  The start point retRef
+ * moves to left and the size retNo increases to match the new chunk.
+ */
+void Dbtup::itdaMergeTabDescr(Uint32& retRef, Uint32& retNo, bool normal)
 {
-   /* THE SIZE OF THE PART TO MERGE MUST BE OF THE SAME SIZE AS THE INSERTED PART */
-   /* THIS IS TRUE EITHER IF ONE PART HAS THE SAME SIZE OR THE SUM OF BOTH PARTS  */
-   /* TOGETHER HAS THE SAME SIZE AS THE PART TO BE INSERTED                       */
-   /* FIND THE SIZES OF THE PARTS TO THE RIGHT OF THE PART TO BE REINSERTED */
+  // merge right
   while ((retRef + retNo) < cnoOfTabDescrRec) {
     ljam();
     Uint32 tabDesRef = retRef + retNo;
@@ -171,11 +169,28 @@ Uint32 Dbtup::itdaMergeTabDescr(Uint32 retRef, Uint32 retNo)
       removeTdArea(tabDesRef, list);
     } else {
       ljam();
-      return retNo;
-    }//if
-  }//while
-  ndbrequire((retRef + retNo) == cnoOfTabDescrRec);
-  return retNo;
+      break;
+    }
+  }
+  // merge left
+  const bool mergeLeft = normal;
+  while (mergeLeft && retRef > 0) {
+    ljam();
+    Uint32 trailerWord = getTabDescrWord(retRef - ZTD_TR_TYPE);
+    if (trailerWord == ZTD_TYPE_FREE) {
+      ljam();
+      Uint32 sizeOfMergedPart = getTabDescrWord(retRef - ZTD_TR_SIZE);
+      ndbrequire(retRef >= sizeOfMergedPart);
+      retRef -= sizeOfMergedPart;
+      retNo += sizeOfMergedPart;
+      Uint32 list = nextHigherTwoLog(sizeOfMergedPart - 1);
+      removeTdArea(retRef, list);
+    } else {
+      ljam();
+      break;
+    }
+  }
+  ndbrequire((retRef + retNo) <= cnoOfTabDescrRec);
 }//Dbtup::itdaMergeTabDescr()
 
 /* ---------------------------------------------------------------- */
@@ -211,3 +226,94 @@ void Dbtup::removeTdArea(Uint32 tabDesRef, Uint32 list)
     setTabDescrWord(tabDescrPrevPtr + ZTD_FL_NEXT, tabDescrNextPtr);
   }//if
 }//Dbtup::removeTdArea()
+
+#ifdef VM_TRACE
+void
+Dbtup::verifytabdes()
+{
+  struct WordType {
+    short fl;   // free list 0-15
+    short ti;   // table id
+    WordType() : fl(-1), ti(-1) {}
+  };
+  WordType* wt = new WordType [cnoOfTabDescrRec];
+  uint free_frags = 0;
+  // free lists
+  {
+    for (uint i = 0; i < 16; i++) {
+      Uint32 desc2 = RNIL;
+      Uint32 desc = cfreeTdList[i];
+      while (desc != RNIL) {
+        const Uint32 size = (1 << i);
+        ndbrequire(size >= ZTD_FREE_SIZE);
+        ndbrequire(desc + size <= cnoOfTabDescrRec);
+        { Uint32 index = desc + ZTD_FL_HEADER;
+          ndbrequire(tableDescriptor[index].tabDescr == ZTD_TYPE_FREE);
+        }
+        { Uint32 index = desc + ZTD_FL_SIZE;
+          ndbrequire(tableDescriptor[index].tabDescr == size);
+        }
+        { Uint32 index = desc + size - ZTD_TR_TYPE;
+          ndbrequire(tableDescriptor[index].tabDescr == ZTD_TYPE_FREE);
+        }
+        { Uint32 index = desc + size - ZTD_TR_SIZE;
+          ndbrequire(tableDescriptor[index].tabDescr == size);
+        }
+        { Uint32 index = desc + ZTD_FL_PREV;
+          ndbrequire(tableDescriptor[index].tabDescr == desc2);
+        }
+        for (uint j = 0; j < size; j++) {
+          ndbrequire(wt[desc + j].fl == -1);
+          wt[desc + j].fl = i;
+        }
+        desc2 = desc;
+        desc = tableDescriptor[desc + ZTD_FL_NEXT].tabDescr;
+        free_frags++;
+      }
+    }
+  }
+  // tables
+  {
+    for (uint i = 0; i < cnoOfTablerec; i++) {
+      TablerecPtr ptr;
+      ptr.i = i;
+      ptrAss(ptr, tablerec);
+      if (ptr.p->tableStatus == DEFINED) {
+        Uint32 offset[10];
+        const Uint32 alloc = getTabDescrOffsets(ptr.p, offset);
+        const Uint32 desc = ptr.p->readKeyArray - offset[3];
+        Uint32 size = alloc;
+        if (size % ZTD_FREE_SIZE != 0)
+          size += ZTD_FREE_SIZE - size % ZTD_FREE_SIZE;
+        ndbrequire(desc + size <= cnoOfTabDescrRec);
+        { Uint32 index = desc + ZTD_FL_HEADER;
+          ndbrequire(tableDescriptor[index].tabDescr == ZTD_TYPE_NORMAL);
+        }
+        { Uint32 index = desc + ZTD_FL_SIZE;
+          ndbrequire(tableDescriptor[index].tabDescr == size);
+        }
+        { Uint32 index = desc + size - ZTD_TR_TYPE;
+          ndbrequire(tableDescriptor[index].tabDescr == ZTD_TYPE_NORMAL);
+        }
+        { Uint32 index = desc + size - ZTD_TR_SIZE;
+          ndbrequire(tableDescriptor[index].tabDescr == size);
+        }
+        for (uint j = 0; j < size; j++) {
+          ndbrequire(wt[desc + j].ti == -1);
+          wt[desc + j].ti = i;
+        }
+      }
+    }
+  }
+  // all words
+  {
+    for (uint i = 0; i < cnoOfTabDescrRec; i++) {
+      bool is_fl = wt[i].fl != -1;
+      bool is_ti = wt[i].ti != -1;
+      ndbrequire(is_fl != is_ti);
+    }
+  }
+  delete [] wt;
+  ndbout << "verifytabdes: frags=" << free_frags << endl;
+}
+#endif
diff --git a/ndb/test/ndbapi/testDict.cpp b/ndb/test/ndbapi/testDict.cpp
index 5f88342705a..3e4ca007978 100644
--- a/ndb/test/ndbapi/testDict.cpp
+++ b/ndb/test/ndbapi/testDict.cpp
@@ -223,6 +223,101 @@ int runCreateAndDrop(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_OK;
 }
 
+int runCreateAndDropAtRandom(NDBT_Context* ctx, NDBT_Step* step)
+{
+  myRandom48Init(NdbTick_CurrentMillisecond());
+  Ndb* pNdb = GETNDB(step);
+  NdbDictionary::Dictionary* pDic = pNdb->getDictionary();
+  int loops = ctx->getNumLoops();
+  int numTables = NDBT_Tables::getNumTables();
+  bool* tabList = new bool [ numTables ];
+  int tabCount;
+
+  {
+    for (int num = 0; num < numTables; num++) {
+      (void)pDic->dropTable(NDBT_Tables::getTable(num)->getName());
+      tabList[num] = false;
+    }
+    tabCount = 0;
+  }
+
+  NdbRestarter restarter;
+  int result = NDBT_OK;
+  int bias = 1; // 0-less 1-more
+  int i = 0;
+  
+  while (i < loops) {
+    g_info << "loop " << i << " tabs " << tabCount << "/" << numTables << endl;
+    int num = myRandom48(numTables);
+    const NdbDictionary::Table* pTab = NDBT_Tables::getTable(num);
+    char tabName[200];
+    strcpy(tabName, pTab->getName());
+
+    if (tabList[num] == false) {
+      if (bias == 0 && myRandom48(100) < 80)
+        continue;
+      g_info << tabName << ": create" << endl;
+      if (pDic->createTable(*pTab) != 0) {
+        const NdbError err = pDic->getNdbError();
+        g_err << tabName << ": create failed: " << err << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      const NdbDictionary::Table* pTab2 = pDic->getTable(tabName);
+      if (pTab2 == NULL) {
+        const NdbError err = pDic->getNdbError();
+        g_err << tabName << ": verify create: " << err << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      tabList[num] = true;
+      assert(tabCount < numTables);
+      tabCount++;
+      if (tabCount == numTables)
+        bias = 0;
+    }
+    else {
+      if (bias == 1 && myRandom48(100) < 80)
+        continue;
+      g_info << tabName << ": drop" << endl;
+      if (restarter.insertErrorInAllNodes(4013) != 0) {
+        g_err << "error insert failed" << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      if (pDic->dropTable(tabName) != 0) {
+        const NdbError err = pDic->getNdbError();
+        g_err << tabName << ": drop failed: " << err << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      const NdbDictionary::Table* pTab2 = pDic->getTable(tabName);
+      if (pTab2 != NULL) {
+        g_err << tabName << ": verify drop: table exists" << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      if (pDic->getNdbError().code != 709 &&
+          pDic->getNdbError().code != 723) {
+        const NdbError err = pDic->getNdbError();
+        g_err << tabName << ": verify drop: " << err << endl;
+        result = NDBT_FAILED;
+        break;
+      }
+      tabList[num] = false;
+      assert(tabCount > 0);
+      tabCount--;
+      if (tabCount == 0)
+        bias = 1;
+    }
+    i++;
+  }
+
+  delete [] tabList;
+  return result;
+}
+
+
 int runCreateAndDropWithData(NDBT_Context* ctx, NDBT_Step* step){
   Ndb* pNdb = GETNDB(step);
   int loops = ctx->getNumLoops();
@@ -1562,6 +1657,12 @@ TESTCASE("CreateAndDrop",
 	 "Try to create and drop the table loop number of times\n"){
   INITIALIZER(runCreateAndDrop);
 }
+TESTCASE("CreateAndDropAtRandom",
+	 "Try to create and drop table at random loop number of times\n"
+         "Uses all available tables\n"
+         "Uses error insert 4013 to make TUP verify table descriptor"){
+  INITIALIZER(runCreateAndDropAtRandom);
+}
 TESTCASE("CreateAndDropWithData", 
 	 "Try to create and drop the table when it's filled with data\n"
 	 "do this loop number of times\n"){
diff --git a/ndb/test/run-test/daily-basic-tests.txt b/ndb/test/run-test/daily-basic-tests.txt
index e70dcafb249..e9064d6e30b 100644
--- a/ndb/test/run-test/daily-basic-tests.txt
+++ b/ndb/test/run-test/daily-basic-tests.txt
@@ -491,6 +491,10 @@ args: -n CreateAndDrop
 
 max-time: 1500
 cmd: testDict
+args: -n CreateAndDropAtRandom -l 200 T1
+
+max-time: 1500
+cmd: testDict
 args: -n CreateAndDropWithData 
 
 max-time: 1500