ndb - bug#20007

  varpages did not get freed on drop table


storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  Add continueb for freeing varpages
storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp:
  Add continueb for freeing varpages
storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp:
  add dropping of varpages
storage/ndb/src/kernel/blocks/dbtup/DbtupPageMap.cpp:
  Add continueb for freeing varpages
storage/ndb/src/kernel/blocks/dbtup/DbtupVarAlloc.cpp:
  Add list of varpage chunks
storage/ndb/src/kernel/blocks/dbtup/tuppage.hpp:
  Add list of varpage chunks

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

@@ -246,6 +246,7 @@ inline const Uint32* ALIGN_WORD(const void* ptr)
 #define ZTUP_SCAN 10
 #define ZFREE_EXTENT 11
 #define ZUNMAP_PAGES 12
+#define ZFREE_VAR_PAGES 13
 
 #define ZSCAN_PROCEDURE 0
 #define ZCOPY_PROCEDURE 2
@@ -620,6 +621,7 @@ struct Fragrecord {
   Uint32 m_tablespace_id;
   Uint32 m_logfile_group_id;
   Disk_alloc_info m_disk_alloc_info;
+  Uint32 m_var_page_chunks;
 };
 typedef Ptr<Fragrecord> FragrecordPtr;
 
@@ -2335,6 +2337,7 @@ private:
 
 
   void releaseFragment(Signal* signal, Uint32 tableId);
+  void drop_fragment_free_var_pages(Signal*);
   void drop_fragment_free_exent(Signal*, TablerecPtr, FragrecordPtr, Uint32);
   void drop_fragment_unmap_pages(Signal*, TablerecPtr, FragrecordPtr, Uint32);
   void drop_fragment_unmap_page_callback(Signal* signal, Uint32, Uint32);

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

@@ -227,6 +227,12 @@ void Dbtup::execCONTINUEB(Signal* signal)
     drop_fragment_unmap_pages(signal, tabPtr, fragPtr, signal->theData[3]);
     return;
   }
+  case ZFREE_VAR_PAGES:
+  {
+    ljam();
+    drop_fragment_free_var_pages(signal);
+    return;
+  }
   default:
     ndbrequire(false);
     break;

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

@@ -140,6 +140,7 @@ void Dbtup::execTUPFRAGREQ(Signal* signal)
   regFragPtr.p->m_undo_complete= false;
   regFragPtr.p->m_lcp_scan_op = RNIL; 
   regFragPtr.p->m_lcp_keep_list = RNIL;
+  regFragPtr.p->m_var_page_chunks = RNIL;
   
   Uint32 noAllocatedPages= allocFragPages(regFragPtr.p, pages);
   
@@ -970,7 +971,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
     case -1:
       break;
     default:
-      ndbrequire(res == pagePtr.i);
+      ndbrequire((Uint32)res == pagePtr.i);
       drop_fragment_unmap_page_callback(signal, pos, res);
     }
     return;
@@ -1052,6 +1053,44 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
     }
   }
   
+  signal->theData[0] = ZFREE_VAR_PAGES;
+  signal->theData[1] = tabPtr.i;
+  signal->theData[2] = fragPtr.i;
+  sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);  
+}
+
+void
+Dbtup::drop_fragment_free_var_pages(Signal* signal)
+{
+  ljam();
+  Uint32 tableId = signal->theData[1];
+  Uint32 fragPtrI = signal->theData[2];
+  
+  TablerecPtr tabPtr;
+  tabPtr.i= tableId;
+  ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
+  
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+  
+  PagePtr pagePtr;
+  if ((pagePtr.i = fragPtr.p->m_var_page_chunks) != RNIL)
+  {
+    c_page_pool.getPtr(pagePtr);
+    Var_page* page = (Var_page*)pagePtr.p;
+    fragPtr.p->m_var_page_chunks = page->next_chunk;
+
+    Uint32 sz = page->chunk_size;
+    returnCommonArea(pagePtr.i, sz);
+    
+    signal->theData[0] = ZFREE_VAR_PAGES;
+    signal->theData[1] = tabPtr.i;
+    signal->theData[2] = fragPtr.i;
+    sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);  
+    return;
+  }
+  
   releaseFragPages(fragPtr.p);
   
   Uint32 i;

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

@@ -299,6 +299,11 @@ void Dbtup::releaseFragPages(Fragrecord* regFragPtr)
 	LocalDLList<Page> tmp(c_page_pool, regFragPtr->thFreeFirst);
 	tmp.remove();
       }
+
+      {
+	LocalSLList<Page> tmp(c_page_pool, regFragPtr->m_empty_pages);
+	tmp.remove();
+      }
       
       return;
     } else {

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

@@ -323,6 +323,13 @@ Dbtup::get_empty_var_page(Fragrecord* fragPtr)
     ptr.p->nextList = RNIL;
     list.add(ret.i + 1, ptr);
   }
+
+  c_page_pool.getPtr(ret);
+  
+  Var_page* page = (Var_page*)ret.p;
+  page->chunk_size = cnt;
+  page->next_chunk = fragPtr->m_var_page_chunks;
+  fragPtr->m_var_page_chunks = ret.i;
   
   return ret.i;
 }

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

@@ -107,8 +107,14 @@ struct Tup_varsize_page
   Uint32 page_state;
   Uint32 next_page;
   Uint32 prev_page;
-  Uint32 first_cluster_page;
-  Uint32 last_cluster_page;
+  union {
+    Uint32 first_cluster_page;
+    Uint32 chunk_size;
+  };
+  union {
+    Uint32 last_cluster_page;
+    Uint32 next_chunk;
+  };
   Uint32 next_cluster_page;
   Uint32 prev_cluster_page;
   Uint32 frag_page_id;