bug#10987 - rpl_ndb_bank

  last bug fix revealed another bug.
  fix buffer management when parsing table descriptor


storage/ndb/src/kernel/blocks/backup/Backup.cpp:
  remove pages on backup record, use buffer in file record instead
storage/ndb/src/kernel/blocks/backup/Backup.hpp:
  remove useless pages on backup record
    file buffer is good enough
storage/ndb/src/kernel/blocks/backup/BackupInit.cpp:
  remove useless pages on backup record
    file buffer is good enough

storage/ndb/src/kernel/blocks/backup/Backup.cpp

@@ -349,7 +349,6 @@ Backup::execDUMP_STATE_ORD(Signal* signal)
       c_backupFilePool.getPtr(lcp_file, lcp.p->dataFilePtr);
       ndbrequire(c_pagePool.getSize() == 
 		 c_pagePool.getNoOfFree() + 
-		 lcp.p->pages.getSize() + 
 		 lcp_file.p->pages.getSize());
     }
   }
@@ -906,7 +905,6 @@ Backup::execBACKUP_REQ(Signal* signal)
     return;
   }//if
 
-  ndbrequire(ptr.p->pages.empty());
   ndbrequire(ptr.p->tables.isEmpty());
   
   ptr.p->m_gsn = 0;
@@ -2756,11 +2754,6 @@ Backup::execGET_TABINFOREF(Signal* signal)
   BackupRecordPtr ptr;
   c_backupPool.getPtr(ptr, senderData);
 
-  ndbout_c("execGET_TABINFOREF: tab: %d errorCode: %d extra: %d",
-	   ref->tableId,
-	   ref->errorCode,
-	   signal->theData[signal->getLength()-1]);
-  
   defineBackupRef(signal, ptr, ref->errorCode);
 }
 
@@ -2791,27 +2784,12 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
   TablePtr tabPtr ;
   ndbrequire(findTable(ptr, tabPtr, tableId));
 
-  /**
-   * No of pages needed
-   */
-  const Uint32 noPages = (len + sizeof(Page32) - 1) / sizeof(Page32);
-  if(ptr.p->pages.getSize() < noPages) {
-    jam();
-    ptr.p->pages.release();
-    if(ptr.p->pages.seize(noPages) == false) {
-      jam();
-      ptr.p->setErrorCode(DefineBackupRef::FailedAllocateTableMem);
-      releaseSections(signal);
-      defineBackupRef(signal, ptr);
-      return;
-    }//if
-  }//if
-  
   BackupFilePtr filePtr;
   ptr.p->files.getPtr(filePtr, ptr.p->ctlFilePtr);
   FsBuffer & buf = filePtr.p->operation.dataBuffer;
+  Uint32* dst = 0;
   { // Write into ctl file
-    Uint32* dst, dstLen = len + 3;
+    Uint32 dstLen = len + 3;
     if(!buf.getWritePtr(&dst, dstLen)) {
       jam();
       ndbrequire(false);
@@ -2835,9 +2813,10 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
     }//if
   }
 
+  releaseSections(signal);
+
   if(ptr.p->checkError()) {
     jam();
-    releaseSections(signal);
     defineBackupRef(signal, ptr);
     return;
   }//if
@@ -2845,7 +2824,6 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
   if (!DictTabInfo::isTable(tabPtr.p->tableType))
   {
     jam();
-    releaseSections(signal);
 
     TablePtr tmp = tabPtr;
     ptr.p->tables.next(tabPtr);
@@ -2853,13 +2831,7 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
     goto next;
   }
   
-  ndbrequire(ptr.p->pages.getSize() >= noPages);
-  Page32Ptr pagePtr;
-  ptr.p->pages.getPtr(pagePtr, 0);
-  copy(&pagePtr.p->data[0], dictTabInfoPtr);
-  releaseSections(signal);
-  
-  if (!parseTableDescription(signal, ptr, tabPtr, len))
+  if (!parseTableDescription(signal, ptr, tabPtr, dst, len))
   {
     jam();
     defineBackupRef(signal, ptr);
@@ -2884,8 +2856,6 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
      */
     jam();
     
-    ptr.p->pages.release();
-    
     if(ptr.p->is_lcp())
     {
       lcp_open_file_done(signal, ptr);
@@ -2916,13 +2886,10 @@ bool
 Backup::parseTableDescription(Signal* signal, 
 			      BackupRecordPtr ptr, 
 			      TablePtr tabPtr, 
+			      const Uint32 * tabdescptr,
 			      Uint32 len)
 {
-
-  Page32Ptr pagePtr;
-  ptr.p->pages.getPtr(pagePtr, 0);
-  
-  SimplePropertiesLinearReader it(&pagePtr.p->data[0], len);
+  SimplePropertiesLinearReader it(tabdescptr, len);
   
   it.first();
   
@@ -4448,7 +4415,6 @@ Backup::cleanup(Signal* signal, BackupRecordPtr ptr)
   ptr.p->files.release();
   ptr.p->tables.release();
   ptr.p->triggers.release();
-  ptr.p->pages.release();
   ptr.p->backupId = ~0;
   
   if(ptr.p->checkError())

storage/ndb/src/kernel/blocks/backup/Backup.hpp

@@ -408,12 +408,12 @@ public:
    * One record per backup
    */
   struct BackupRecord {
-    BackupRecord(Backup& b, ArrayPool<Page32> & pp, 
+    BackupRecord(Backup& b, 
 		 ArrayPool<Table> & tp, 
 		 ArrayPool<BackupFile> & bp,
 		 ArrayPool<TriggerRecord> & trp) 
       : slaveState(b, validSlaveTransitions, validSlaveTransitionsCount,1)
-      , tables(tp), triggers(trp), files(bp), pages(pp)
+      , tables(tp), triggers(trp), files(bp)
       , masterData(b), backup(b)
       {
       }
@@ -448,7 +448,6 @@ public:
     Uint32 dataFilePtr; // Ptr.i to first data-file
     
     Uint32 backupDataLen;  // Used for (un)packing backup request
-    Array<Page32> pages;   // Used for (un)packing backup request
     SimpleProperties props;// Used for (un)packing backup request
 
     struct SlaveData {
@@ -608,7 +607,7 @@ public:
 
   NodeId getMasterNodeId() const { return c_masterNodeId; }
   bool findTable(const BackupRecordPtr &, TablePtr &, Uint32 tableId) const;
-  bool parseTableDescription(Signal*, BackupRecordPtr ptr, TablePtr, Uint32);
+  bool parseTableDescription(Signal*, BackupRecordPtr ptr, TablePtr, const Uint32*, Uint32);
   
   bool insertFileHeader(BackupFormat::FileType, BackupRecord*, BackupFile*);
   void sendBackupRef(Signal* signal, BackupRecordPtr ptr, Uint32 errorCode);

storage/ndb/src/kernel/blocks/backup/BackupInit.cpp

@@ -206,7 +206,7 @@ Backup::execREAD_CONFIG_REQ(Signal* signal)
     ArrayList<BackupRecord> recs(c_backupPool);
     BackupRecordPtr ptr;
     while(recs.seize(ptr)){
-      new (ptr.p) BackupRecord(* this, c_pagePool, c_tablePool, 
+      new (ptr.p) BackupRecord(* this, c_tablePool, 
 			       c_backupFilePool, c_triggerPool);
     }
     recs.release();