Fix for Bug #18413 Data usage for varsize columns are not correctly reported to mysqld

sql/ha_ndbcluster.cc

@@ -7420,7 +7420,7 @@ ndb_get_table_statistics(Ndb* ndb, const NDBTAB *ndbtab,
 
   do
   {
-    Uint64 rows, commits, mem;
+    Uint64 rows, commits, fixed_mem, var_mem;
     Uint32 size;
     Uint32 count= 0;
     Uint64 sum_rows= 0;
@@ -7458,7 +7458,10 @@ ndb_get_table_statistics(Ndb* ndb, const NDBTAB *ndbtab,
     pOp->getValue(NdbDictionary::Column::ROW_COUNT, (char*)&rows);
     pOp->getValue(NdbDictionary::Column::COMMIT_COUNT, (char*)&commits);
     pOp->getValue(NdbDictionary::Column::ROW_SIZE, (char*)&size);
-    pOp->getValue(NdbDictionary::Column::FRAGMENT_MEMORY, (char*)&mem);
+    pOp->getValue(NdbDictionary::Column::FRAGMENT_FIXED_MEMORY, 
+		  (char*)&fixed_mem);
+    pOp->getValue(NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY, 
+		  (char*)&var_mem);
     
     if (pTrans->execute(NdbTransaction::NoCommit,
                         NdbTransaction::AbortOnError,
@@ -7474,7 +7477,7 @@ ndb_get_table_statistics(Ndb* ndb, const NDBTAB *ndbtab,
       sum_commits+= commits;
       if (sum_row_size < size)
         sum_row_size= size;
-      sum_mem+= mem;
+      sum_mem+= fixed_mem + var_mem;
       count++;
     }
     

storage/ndb/include/kernel/AttributeHeader.hpp

@@ -39,12 +39,13 @@ public:
   STATIC_CONST( RANGE_NO     = 0xFFFB ); // Read range no (when batched ranges)
   
   STATIC_CONST( ROW_SIZE     = 0xFFFA );
-  STATIC_CONST( FRAGMENT_MEMORY= 0xFFF9 );
+  STATIC_CONST( FRAGMENT_FIXED_MEMORY= 0xFFF9 );
 
   STATIC_CONST( RECORDS_IN_RANGE = 0xFFF8 );
   STATIC_CONST( DISK_REF     = 0xFFF7 );
   STATIC_CONST( ROWID        = 0xFFF6 );
   STATIC_CONST( ROW_GCI      = 0xFFF5 );
+  STATIC_CONST( FRAGMENT_VARSIZED_MEMORY = 0xFFF4 );
 
   // NOTE: in 5.1 ctors and init take size in bytes
 

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -525,7 +525,8 @@ public:
     const char* getDefaultValue() const;
 
     static const Column * FRAGMENT;
-    static const Column * FRAGMENT_MEMORY;
+    static const Column * FRAGMENT_FIXED_MEMORY;
+    static const Column * FRAGMENT_VARSIZED_MEMORY;
     static const Column * ROW_COUNT;
     static const Column * COMMIT_COUNT;
     static const Column * ROW_SIZE;

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

@@ -604,6 +604,7 @@ struct Fragrecord {
   Uint32 currentPageRange;
   Uint32 rootPageRange;
   Uint32 noOfPages;
+  Uint32 noOfVarPages;
   Uint32 noOfPagesToGrow;
 
   DLList<Page>::Head emptyPrimPage; // allocated pages (not init)

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

@@ -351,6 +351,7 @@ void Dbtup::initFragRange(Fragrecord* const regFragPtr)
   regFragPtr->rootPageRange = RNIL;
   regFragPtr->currentPageRange = RNIL;
   regFragPtr->noOfPages = 0;
+  regFragPtr->noOfVarPages = 0;
   regFragPtr->noOfPagesToGrow = 2;
   regFragPtr->nextStartRange = 0;
 }//initFragRange()

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

@@ -1135,13 +1135,20 @@ Dbtup::read_pseudo(Uint32 attrId,
   case AttributeHeader::FRAGMENT:
     * outBuffer = fragptr.p->fragmentId;
     return 1;
-  case AttributeHeader::FRAGMENT_MEMORY:
+  case AttributeHeader::FRAGMENT_FIXED_MEMORY:
   {
     Uint64 tmp= fragptr.p->noOfPages;
     tmp*= 32768;
     memcpy(outBuffer,&tmp,8);
   }
   return 2;
+  case AttributeHeader::FRAGMENT_VARSIZED_MEMORY:
+  {
+    Uint64 tmp= fragptr.p->noOfVarPages;
+    tmp*= 32768;
+    memcpy(outBuffer,&tmp,8);
+  }
+  return 2;
   case AttributeHeader::ROW_SIZE:
     * outBuffer = tabptr.p->m_offsets[MM].m_fix_header_size << 2;
     return 1;

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

@@ -302,6 +302,7 @@ Dbtup::get_empty_var_page(Fragrecord* fragPtr)
 
   Uint32 cnt;
   allocConsPages(10, cnt, ptr.i);
+  fragPtr->noOfVarPages+= cnt;
   if (unlikely(cnt == 0))
   {
     return RNIL;

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -328,9 +328,14 @@ NdbColumnImpl::create_pseudo(const char * name){
     col->m_impl.m_attrId = AttributeHeader::FRAGMENT;
     col->m_impl.m_attrSize = 4;
     col->m_impl.m_arraySize = 1;
-  } else if(!strcmp(name, "NDB$FRAGMENT_MEMORY")){
+  } else if(!strcmp(name, "NDB$FRAGMENT_FIXED_MEMORY")){
     col->setType(NdbDictionary::Column::Bigunsigned);
-    col->m_impl.m_attrId = AttributeHeader::FRAGMENT_MEMORY;
+    col->m_impl.m_attrId = AttributeHeader::FRAGMENT_FIXED_MEMORY;
+    col->m_impl.m_attrSize = 8;
+    col->m_impl.m_arraySize = 1;
+  } else if(!strcmp(name, "NDB$FRAGMENT_VARSIZED_MEMORY")){
+    col->setType(NdbDictionary::Column::Bigunsigned);
+    col->m_impl.m_attrId = AttributeHeader::FRAGMENT_VARSIZED_MEMORY;
     col->m_impl.m_attrSize = 8;
     col->m_impl.m_arraySize = 1;
   } else if(!strcmp(name, "NDB$ROW_COUNT")){
@@ -1316,7 +1321,8 @@ NdbDictionaryImpl::~NdbDictionaryImpl()
     m_globalHash->lock();
     if(--f_dictionary_count == 0){
       delete NdbDictionary::Column::FRAGMENT; 
-      delete NdbDictionary::Column::FRAGMENT_MEMORY;
+      delete NdbDictionary::Column::FRAGMENT_FIXED_MEMORY;
+      delete NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY;
       delete NdbDictionary::Column::ROW_COUNT;
       delete NdbDictionary::Column::COMMIT_COUNT;
       delete NdbDictionary::Column::ROW_SIZE;
@@ -1326,7 +1332,8 @@ NdbDictionaryImpl::~NdbDictionaryImpl()
       delete NdbDictionary::Column::ROWID;
       delete NdbDictionary::Column::ROW_GCI;
       NdbDictionary::Column::FRAGMENT= 0;
-      NdbDictionary::Column::FRAGMENT_MEMORY= 0;
+      NdbDictionary::Column::FRAGMENT_FIXED_MEMORY= 0;
+      NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY= 0;
       NdbDictionary::Column::ROW_COUNT= 0;
       NdbDictionary::Column::COMMIT_COUNT= 0;
       NdbDictionary::Column::ROW_SIZE= 0;
@@ -1483,8 +1490,10 @@ NdbDictionaryImpl::setTransporter(class Ndb* ndb,
     if(f_dictionary_count++ == 0){
       NdbDictionary::Column::FRAGMENT= 
 	NdbColumnImpl::create_pseudo("NDB$FRAGMENT");
-      NdbDictionary::Column::FRAGMENT_MEMORY= 
-	NdbColumnImpl::create_pseudo("NDB$FRAGMENT_MEMORY");
+      NdbDictionary::Column::FRAGMENT_FIXED_MEMORY= 
+	NdbColumnImpl::create_pseudo("NDB$FRAGMENT_FIXED_MEMORY");
+      NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY= 
+	NdbColumnImpl::create_pseudo("NDB$FRAGMENT_VARSIZED_MEMORY");
       NdbDictionary::Column::ROW_COUNT= 
 	NdbColumnImpl::create_pseudo("NDB$ROW_COUNT");
       NdbDictionary::Column::COMMIT_COUNT= 
@@ -5041,7 +5050,8 @@ template class Vector<NdbTableImpl*>;
 template class Vector<NdbColumnImpl*>;
 
 const NdbDictionary::Column * NdbDictionary::Column::FRAGMENT = 0;
-const NdbDictionary::Column * NdbDictionary::Column::FRAGMENT_MEMORY = 0;
+const NdbDictionary::Column * NdbDictionary::Column::FRAGMENT_FIXED_MEMORY = 0;
+const NdbDictionary::Column * NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY = 0;
 const NdbDictionary::Column * NdbDictionary::Column::ROW_COUNT = 0;
 const NdbDictionary::Column * NdbDictionary::Column::COMMIT_COUNT = 0;
 const NdbDictionary::Column * NdbDictionary::Column::ROW_SIZE = 0;

storage/ndb/tools/desc.cpp

@@ -293,7 +293,8 @@ void print_part_info(Ndb* pNdb, NDBT_Table* pTab)
     { "Partition", 0, NdbDictionary::Column::FRAGMENT },
     { "Row count", 0, NdbDictionary::Column::ROW_COUNT },
     { "Commit count", 0, NdbDictionary::Column::COMMIT_COUNT },
-    { "Frag memory", 0, NdbDictionary::Column::FRAGMENT_MEMORY },
+    { "Frag fixed memory", 0, NdbDictionary::Column::FRAGMENT_FIXED_MEMORY },
+    { "Frag varsized memory", 0, NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY },
     { 0, 0, 0 }
   };