Merge ts3-168.ts.cn.tlan:/home/ngb/mysql/mysql-5.1/mysql-5.1-new-ndb-bj

into  ts3-168.ts.cn.tlan:/home/ngb/mysql/mysql-5.1/mysql-5.1-new-ndb

mysys/my_open.c

@@ -167,9 +167,17 @@ File my_register_filename(File fd, const char *FileName, enum file_type
   else
     my_errno=errno;
   DBUG_PRINT("error",("Got error %d on open",my_errno));
-  if (MyFlags & (MY_FFNF | MY_FAE | MY_WME))
+  if (MyFlags & (MY_FFNF | MY_FAE | MY_WME)) {
+    if (my_errno == EMFILE) {
+      DBUG_PRINT("error",("print err: %d",EE_OUT_OF_FILERESOURCES));
+      my_error(EE_OUT_OF_FILERESOURCES, MYF(ME_BELL+ME_WAITTANG),
+	     FileName, my_errno);
+    } else {
+      DBUG_PRINT("error",("print err: %d",error_message_number));
       my_error(error_message_number, MYF(ME_BELL+ME_WAITTANG),
 	     FileName, my_errno);
+    }
+  }
   return(fd);
 }
 

sql/ha_ndbcluster.cc

@@ -4825,10 +4825,18 @@ int ha_ndbcluster::create(const char *name,
   }
 
   if (info->store_on_disk)
+  { 
     if (info->tablespace)
       tab.setTablespace(info->tablespace);
     else
       tab.setTablespace("DEFAULT-TS");
+  }
+  else if (info->tablespace)
+  {
+    tab.setTablespace(info->tablespace);
+    info->store_on_disk = true;  //if use tablespace, that also means store on disk
+  }
+
   // No primary key, create shadow key as 64 bit, auto increment  
   if (form->s->primary_key == MAX_KEY) 
   {
@@ -6537,6 +6545,23 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
   
   hash_free(&ok_tables);
   hash_free(&ndb_tables);
+
+  // Delete schema file from files
+  if (!strcmp(db, NDB_REP_DB))
+  {
+    uint count = 0;
+    while (count++ < files->elements)
+    {
+      file_name = (char *)files->pop();
+      if (!strcmp(file_name, NDB_SCHEMA_TABLE))
+      {
+        DBUG_PRINT("info", ("skip %s.%s table, it should be hidden to user",
+                   NDB_REP_DB, NDB_SCHEMA_TABLE));
+        continue;
+      }
+      files->push_back(file_name); 
+    }
+  }
   } // extra bracket to avoid gcc 2.95.3 warning
   DBUG_RETURN(0);    
 }

sql/log.cc

@@ -2973,6 +2973,7 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
                           ulonglong *decrease_log_space)
 {
   int error;
+  int ret = 0;
   bool exit_loop= 0;
   LOG_INFO log_info;
   DBUG_ENTER("purge_logs");
@@ -3017,6 +3018,14 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
       *decrease_log_space-= file_size;
 
     ha_binlog_index_purge_file(current_thd, log_info.log_file_name);
+    if (current_thd->query_error) {
+      DBUG_PRINT("info",("query error: %d", current_thd->query_error));
+      if (my_errno == EMFILE) {
+        DBUG_PRINT("info",("my_errno: %d, set ret = LOG_INFO_EMFILE", my_errno));
+        ret = LOG_INFO_EMFILE;
+        break;
+      }
+    }
 
     if (find_next_log(&log_info, 0) || exit_loop)
       break;
@@ -3027,6 +3036,9 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
     the log index file after restart - otherwise, this should be safe
   */
   error= update_log_index(&log_info, need_update_threads);
+  if (error == 0) {
+    error = ret;
+  }
 
 err:
   if (need_mutex)

sql/log.h

@@ -114,6 +114,8 @@ extern TC_LOG_DUMMY tc_log_dummy;
 #define LOG_INFO_MEM -6
 #define LOG_INFO_FATAL -7
 #define LOG_INFO_IN_USE -8
+#define LOG_INFO_EMFILE -9
+
 
 /* bitmap to SQL_LOG::close() */
 #define LOG_CLOSE_INDEX		1

sql/share/errmsg.txt

@@ -6012,4 +6012,5 @@ ER_WRONG_PARAMETERS_TO_NATIVE_FCT 42000
         eng "Incorrect parameters in the call to native function '%-.64s'"
 ER_NATIVE_FCT_NAME_COLLISION
         eng "This function '%-.64s' has the same name as a native function."
-
+ER_BINLOG_PURGE_EMFILE
+        eng "Too many files opened, please execute the command again"

sql/sql_repl.cc

@@ -239,6 +239,7 @@ bool purge_error_message(THD* thd, int res)
   case LOG_INFO_MEM:	errmsg= ER_OUT_OF_RESOURCES; break;
   case LOG_INFO_FATAL:	errmsg= ER_BINLOG_PURGE_FATAL_ERR; break;
   case LOG_INFO_IN_USE: errmsg= ER_LOG_IN_USE; break;
+  case LOG_INFO_EMFILE: errmsg= ER_BINLOG_PURGE_EMFILE; break;
   default:		errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;
   }
 

sql/table.cc

@@ -1562,21 +1562,32 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
                        outparam->file->auto_repair() &&
                        !(ha_open_flags & HA_OPEN_FOR_REPAIR));
 
-      if (ha_err == HA_ERR_NO_SUCH_TABLE)
+      switch (ha_err)
       {
+        case HA_ERR_NO_SUCH_TABLE:
 	  /*
             The table did not exists in storage engine, use same error message
             as if the .frm file didn't exist
           */
 	  error= 1;
 	  my_errno= ENOENT;
-      }
-      else
-      {
+          break;
+        case EMFILE:
+	  /*
+            Too many files opened, use same error message as if the .frm
+            file can't open
+           */
+          DBUG_PRINT("error", ("open file: %s failed, too many files opened (errno: %d)", 
+		  share->normalized_path.str, ha_err));
+	  error= 1;
+	  my_errno= EMFILE;
+          break;
+        default:
           outparam->file->print_error(ha_err, MYF(0));
           error_reported= TRUE;
           if (ha_err == HA_ERR_TABLE_DEF_CHANGED)
             error= 7;
+          break;
       }
       goto err;                                 /* purecov: inspected */
     }

storage/ndb/include/mgmapi/mgmapi.h

@@ -1182,6 +1182,14 @@ extern "C" {
   int ndb_mgm_check_connection(NdbMgmHandle handle);
 
   int ndb_mgm_report_event(NdbMgmHandle handle, Uint32 *data, Uint32 length);
+
+  struct ndb_mgm_param_info
+  {
+    Uint32 m_id;
+    const char * m_name;
+  };
+  int ndb_mgm_get_db_parameter_info(Uint32 paramId, struct ndb_mgm_param_info * info, 
+             size_t * size);
 #endif
 
 #ifndef DOXYGEN_SHOULD_SKIP_DEPRECATED

storage/ndb/include/ndbapi/ndberror.h

@@ -99,7 +99,7 @@ typedef  ndberror_classification_enum ndberror_classification;
 const char *ndberror_status_message(ndberror_status);
 const char *ndberror_classification_message(ndberror_classification);
 void ndberror_update(ndberror_struct *);
-int ndb_error_string(int err_no, char *str, unsigned int size);
+int ndb_error_string(int err_no, char *str, int size);
 
 #endif /* doxygen skip internal*/
 

storage/ndb/src/common/transporter/TCP_Transporter.cpp

@@ -20,6 +20,9 @@
 #include "TCP_Transporter.hpp"
 #include <NdbOut.hpp>
 #include <NdbSleep.h>
+
+#include <EventLogger.hpp>
+extern EventLogger g_eventLogger;
 // End of stuff to be moved
 
 #ifdef NDB_WIN32
@@ -153,14 +156,14 @@ TCP_Transporter::setSocketOptions(){
   if (setsockopt(theSocket, SOL_SOCKET, SO_RCVBUF,
                  (char*)&sockOptRcvBufSize, sizeof(sockOptRcvBufSize)) < 0) {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("The setsockopt SO_RCVBUF error code = %d", InetErrno);
+    g_eventLogger.error("The setsockopt SO_RCVBUF error code = %d", InetErrno);
 #endif
   }//if
   
   if (setsockopt(theSocket, SOL_SOCKET, SO_SNDBUF,
                  (char*)&sockOptSndBufSize, sizeof(sockOptSndBufSize)) < 0) {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("The setsockopt SO_SNDBUF error code = %d", InetErrno);
+    g_eventLogger.error("The setsockopt SO_SNDBUF error code = %d", InetErrno);
 #endif
   }//if
   
@@ -171,7 +174,7 @@ TCP_Transporter::setSocketOptions(){
   if (setsockopt(theSocket, IPPROTO_TCP, TCP_NODELAY, 
                  (char*)&sockOptNodelay, sizeof(sockOptNodelay)) < 0) {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("The setsockopt TCP_NODELAY error code = %d", InetErrno);
+    g_eventLogger.error("The setsockopt TCP_NODELAY error code = %d", InetErrno);
 #endif
   }//if
 }
@@ -185,7 +188,7 @@ TCP_Transporter::setSocketNonBlocking(NDB_SOCKET_TYPE socket){
   if(ioctlsocket(socket, FIONBIO, &ul))
   {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("Set non-blocking server error3: %d", InetErrno);
+    g_eventLogger.error("Set non-blocking server error3: %d", InetErrno);
 #endif
   }//if
   return true;
@@ -199,13 +202,13 @@ TCP_Transporter::setSocketNonBlocking(NDB_SOCKET_TYPE socket){
   flags = fcntl(socket, F_GETFL, 0);
   if (flags < 0) {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("Set non-blocking server error1: %s", strerror(InetErrno));
+    g_eventLogger.error("Set non-blocking server error1: %s", strerror(InetErrno));
 #endif
   }//if
   flags |= NDB_NONBLOCK;
   if (fcntl(socket, F_SETFL, flags) == -1) {
 #ifdef DEBUG_TRANSPORTER
-    ndbout_c("Set non-blocking server error2: %s", strerror(InetErrno));
+    g_eventLogger.error("Set non-blocking server error2: %s", strerror(InetErrno));
 #endif
   }//if
   return true;
@@ -326,7 +329,7 @@ TCP_Transporter::doSend() {
     } else {
       // Send failed
 #if defined DEBUG_TRANSPORTER
-      ndbout_c("Send Failure(disconnect==%d) to node = %d nBytesSent = %d "
+      g_eventLogger.error("Send Failure(disconnect==%d) to node = %d nBytesSent = %d "
 	       "errno = %d strerror = %s",
 	       DISCONNECT_ERRNO(InetErrno, nBytesSent),
 	       remoteNodeId, nBytesSent, InetErrno, 
@@ -361,11 +364,11 @@ TCP_Transporter::doReceive() {
       
       if(receiveBuffer.sizeOfData > receiveBuffer.sizeOfBuffer){
 #ifdef DEBUG_TRANSPORTER
-	ndbout_c("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
+	g_eventLogger.error("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
 		 receiveBuffer.sizeOfData, receiveBuffer.sizeOfBuffer);
-	ndbout_c("nBytesRead = %d", nBytesRead);
+	g_eventLogger.error("nBytesRead = %d", nBytesRead);
 #endif
-	ndbout_c("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
+	g_eventLogger.error("receiveBuffer.sizeOfData(%d) > receiveBuffer.sizeOfBuffer(%d)",
 		 receiveBuffer.sizeOfData, receiveBuffer.sizeOfBuffer);
 	report_error(TE_INVALID_MESSAGE_LENGTH);
 	return 0;
@@ -382,7 +385,7 @@ TCP_Transporter::doReceive() {
       return nBytesRead;
     } else {
 #if defined DEBUG_TRANSPORTER
-      ndbout_c("Receive Failure(disconnect==%d) to node = %d nBytesSent = %d "
+      g_eventLogger.error("Receive Failure(disconnect==%d) to node = %d nBytesSent = %d "
 	       "errno = %d strerror = %s",
 	       DISCONNECT_ERRNO(InetErrno, nBytesRead),
 	       remoteNodeId, nBytesRead, InetErrno, 

storage/ndb/src/common/transporter/TransporterRegistry.cpp

@@ -778,7 +778,7 @@ TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
   
   tcpReadSelectReply = select(maxSocketValue, &tcpReadset, 0, 0, &timeout);  
   if(false && tcpReadSelectReply == -1 && errno == EINTR)
-    ndbout_c("woke-up by signal");
+    g_eventLogger.info("woke-up by signal");
 
 #ifdef NDB_WIN32
   if(tcpReadSelectReply == SOCKET_ERROR)
@@ -1112,12 +1112,12 @@ TransporterRegistry::start_clients_thread()
 	      }
 	      else if(ndb_mgm_is_connected(m_mgm_handle))
 	      {
-		ndbout_c("Failed to get dynamic port to connect to: %d", res);
+		g_eventLogger.info("Failed to get dynamic port to connect to: %d", res);
 		ndb_mgm_disconnect(m_mgm_handle);
 	      }
 	      else
 	      {
-		ndbout_c("Management server closed connection early. "
+		g_eventLogger.info("Management server closed connection early. "
 			 "It is probably being shut down (or has problems). "
 			 "We will retry the connection.");
 	      }
@@ -1215,7 +1215,7 @@ TransporterRegistry::start_service(SocketServer& socket_server)
   DBUG_ENTER("TransporterRegistry::start_service");
   if (m_transporter_interface.size() > 0 && !nodeIdSpecified)
   {
-    ndbout_c("TransporterRegistry::startReceiving: localNodeId not specified");
+    g_eventLogger.error("TransporterRegistry::startReceiving: localNodeId not specified");
     DBUG_RETURN(false);
   }
 
@@ -1241,7 +1241,7 @@ TransporterRegistry::start_service(SocketServer& socket_server)
 	 * If it wasn't a dynamically allocated port, or
 	 * our attempts at getting a new dynamic port failed
 	 */
-	ndbout_c("Unable to setup transporter service port: %s:%d!\n"
+	g_eventLogger.error("Unable to setup transporter service port: %s:%d!\n"
 		 "Please check if the port is already used,\n"
 		 "(perhaps the node is already running)",
 		 t.m_interface ? t.m_interface : "*", t.m_s_service_port);
@@ -1338,13 +1338,13 @@ bool TransporterRegistry::connect_client(NdbMgmHandle *h)
 
   if(!mgm_nodeid)
   {
-    ndbout_c("%s: %d", __FILE__, __LINE__);
+    g_eventLogger.error("%s: %d", __FILE__, __LINE__);
     return false;
   }
   Transporter * t = theTransporters[mgm_nodeid];
   if (!t)
   {
-    ndbout_c("%s: %d", __FILE__, __LINE__);
+    g_eventLogger.error("%s: %d", __FILE__, __LINE__);
     return false;
   }
   DBUG_RETURN(t->connect_client(connect_ndb_mgmd(h)));
@@ -1360,7 +1360,7 @@ NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(NdbMgmHandle *h)
 
   if ( h==NULL || *h == NULL )
   {
-    ndbout_c("%s: %d", __FILE__, __LINE__);
+    g_eventLogger.error("%s: %d", __FILE__, __LINE__);
     return NDB_INVALID_SOCKET;
   }
 
@@ -1373,10 +1373,10 @@ NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(NdbMgmHandle *h)
 				   m_transporter_interface[i].m_s_service_port,
 				   &mgm_reply) < 0)
     {
-      ndbout_c("Error: %s: %d",
+      g_eventLogger.error("Error: %s: %d",
 	       ndb_mgm_get_latest_error_desc(*h),
 	       ndb_mgm_get_latest_error(*h));
-      ndbout_c("%s: %d", __FILE__, __LINE__);
+      g_eventLogger.error("%s: %d", __FILE__, __LINE__);
       ndb_mgm_destroy_handle(h);
       return NDB_INVALID_SOCKET;
     }
@@ -1388,10 +1388,10 @@ NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(NdbMgmHandle *h)
   NDB_SOCKET_TYPE sockfd= ndb_mgm_convert_to_transporter(h);
   if ( sockfd == NDB_INVALID_SOCKET)
   {
-    ndbout_c("Error: %s: %d",
+    g_eventLogger.error("Error: %s: %d",
 	     ndb_mgm_get_latest_error_desc(*h),
 	     ndb_mgm_get_latest_error(*h));
-    ndbout_c("%s: %d", __FILE__, __LINE__);
+    g_eventLogger.error("%s: %d", __FILE__, __LINE__);
     ndb_mgm_destroy_handle(h);
   }
   return sockfd;

storage/ndb/src/common/util/File.cpp

@@ -122,12 +122,24 @@ bool
 File_class::close()
 {
   bool rc = true;
+  int retval = 0;
+
   if (m_file != NULL)
   { 
     ::fflush(m_file);
-    rc = (::fclose(m_file) == 0 ? true : false);
-    m_file = NULL; // Try again?
+    retval = ::fclose(m_file);
+    while ( (retval != 0) && (errno == EINTR) ){
+      retval = ::fclose(m_file);
+    }
+    if( retval == 0){
+      rc = true;
+    }
+    else {
+      rc = false;
+      ndbout_c("ERROR: Close file error in File.cpp for %s",strerror(errno));
+    }   
   }  
+  m_file = NULL;
   
   return rc;
 }

storage/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp

@@ -62,7 +62,8 @@ Cmvmi::Cmvmi(Block_context& ctx) :
 			    &long_sig_buffer_size);
 
   long_sig_buffer_size= long_sig_buffer_size / 256;
-  g_sectionSegmentPool.setSize(long_sig_buffer_size);
+  g_sectionSegmentPool.setSize(long_sig_buffer_size,
+                               false,true,true,CFG_DB_LONG_SIGNAL_BUFFER);
 
   // Add received signals
   addRecSignal(GSN_CONNECT_REP, &Cmvmi::execCONNECT_REP);

storage/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp

@@ -52,7 +52,8 @@ void Dbacc::initRecords()
   page8 = (Page8*)allocRecord("Page8",
 			      sizeof(Page8), 
 			      cpagesize,
-			      false);
+			      false,
+            CFG_DB_INDEX_MEM);
 
   operationrec = (Operationrec*)allocRecord("Operationrec",
 					    sizeof(Operationrec),

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

@@ -311,7 +311,7 @@ void Dbtup::execREAD_CONFIG_REQ(Signal* signal)
 
   c_storedProcPool.setSize(noOfStoredProc);
   c_buildIndexPool.setSize(c_noOfBuildIndexRec);
-  c_triggerPool.setSize(noOfTriggers);
+  c_triggerPool.setSize(noOfTriggers, false, true, true, CFG_DB_NO_TRIGGERS);
 
   c_extent_hash.setSize(1024); // 4k
   
@@ -343,6 +343,7 @@ void Dbtup::initRecords()
 {
   unsigned i;
   Uint32 tmp;
+  Uint32 tmp1 = 0;
   const ndb_mgm_configuration_iterator * p = 
     m_ctx.m_config.getOwnConfigIterator();
   ndbrequire(p != 0);
@@ -350,7 +351,7 @@ void Dbtup::initRecords()
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUP_PAGE, &tmp));
 
   // Records with dynamic sizes
-  Page* ptr =(Page*)allocRecord("Page", sizeof(Page), tmp, false);
+  Page* ptr =(Page*)allocRecord("Page", sizeof(Page), tmp, false, CFG_DB_DATA_MEM);
   c_page_pool.set(ptr, tmp);
   
   attrbufrec = (Attrbufrec*)allocRecord("Attrbufrec", 
@@ -374,7 +375,9 @@ void Dbtup::initRecords()
 						  cnoOfTabDescrRec);
 
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUP_OP_RECS, &tmp));
-  c_operation_pool.setSize(tmp);
+  ndb_mgm_get_int_parameter(p, CFG_DB_NO_LOCAL_OPS, &tmp1);
+  c_operation_pool.setSize(tmp, false, true, true, 
+      tmp1 == 0 ? CFG_DB_NO_OPS : CFG_DB_NO_LOCAL_OPS);
   
   pageRange = (PageRange*)allocRecord("PageRange",
 				      sizeof(PageRange), 

storage/ndb/src/kernel/vm/ArrayPool.hpp

@@ -25,6 +25,7 @@
 #include <ErrorReporter.hpp>
 #include <NdbMem.h>
 #include <Bitmask.hpp>
+#include <mgmapi.h>
 
 template <class T> class Array;
 
@@ -43,7 +44,8 @@ public:
    *
    * Note, can currently only be called once
    */
-  bool setSize(Uint32 noOfElements, bool align = false, bool exit_on_error = true, bool guard = true);
+  bool setSize(Uint32 noOfElements, bool align = false, bool exit_on_error = true, 
+               bool guard = true, Uint32 paramId = 0);
   bool set(T*, Uint32 cnt, bool align = false);
   void clear() { theArray = 0; }
 
@@ -221,13 +223,19 @@ template <class T>
 inline
 bool
 ArrayPool<T>::setSize(Uint32 noOfElements, 
-		      bool align, bool exit_on_error, bool guard){
+		      bool align, bool exit_on_error, bool guard, Uint32 paramId){
   if(size == 0){
     if(noOfElements == 0)
       return true;
+  Uint64 real_size = (Uint64)noOfElements * sizeof(T);
+  size_t req_size = (size_t)real_size;
+  Uint64 real_size_align = real_size + sizeof(T);
+  size_t req_size_align = (size_t)real_size_align;
+
     if(align)
     {
-      alloc_ptr = ndbd_malloc((noOfElements+1) * sizeof(T));
+      if((Uint64)req_size_align == real_size_align && req_size_align > 0)
+        alloc_ptr = ndbd_malloc(req_size_align);  
       UintPtr p = (UintPtr)alloc_ptr;
       UintPtr mod = p % sizeof(T);
       if (mod)
@@ -236,14 +244,23 @@ ArrayPool<T>::setSize(Uint32 noOfElements,
       }
       theArray = (T *)p;
     }
-    else
-      theArray = (T *)(alloc_ptr = ndbd_malloc(noOfElements * sizeof(T)));
+    else if((Uint64)req_size == real_size && req_size > 0)
+      theArray = (T *)(alloc_ptr = ndbd_malloc(req_size));
 
     if(theArray == 0)
     {
+      char errmsg[255] = "ArrayPool<T>::setSize malloc failed";
+      struct ndb_mgm_param_info param_info;
+      size_t size = sizeof(ndb_mgm_param_info);
       if (!exit_on_error)
 	return false;
-      ErrorReporter::handleAssert("ArrayPool<T>::setSize malloc failed",
+
+      if(0 != paramId && 0 == ndb_mgm_get_db_parameter_info(paramId, &param_info, &size)) {
+        BaseString::snprintf(errmsg, sizeof(errmsg), 
+                "Malloc memory for %s failed", param_info.m_name);
+      }
+
+      ErrorReporter::handleAssert(errmsg,
 				  __FILE__, __LINE__, NDBD_EXIT_MEMALLOC);
       return false; // not reached
     }

storage/ndb/src/kernel/vm/SimulatedBlock.cpp

@@ -657,29 +657,39 @@ SimulatedBlock::getBatSize(Uint16 blockNo){
 }
 
 void* 
-SimulatedBlock::allocRecord(const char * type, size_t s, size_t n, bool clear) 
+SimulatedBlock::allocRecord(const char * type, size_t s, size_t n, bool clear, Uint32 paramId) 
 {
 
   void * p = NULL;
   size_t size = n*s;
+  Uint64 real_size = (Uint64)((Uint64)n)*((Uint64)s);
   refresh_watch_dog(); 
-  if (size > 0){
+  if (real_size > 0){
 #ifdef VM_TRACE_MEM
-    ndbout_c("%s::allocRecord(%s, %u, %u) = %u bytes", 
+    ndbout_c("%s::allocRecord(%s, %u, %u) = %llu bytes", 
 	     getBlockName(number()), 
 	     type,
 	     s,
 	     n,
-	     size);
+	     real_size);
 #endif
+    if( real_size == (Uint64)size )
       p = ndbd_malloc(size);
     if (p == NULL){
       char buf1[255];
       char buf2[255];
+      struct ndb_mgm_param_info param_info;
+      size_t size = sizeof(ndb_mgm_param_info);
+
+      if(0 != paramId && 0 == ndb_mgm_get_db_parameter_info(paramId, &param_info, &size)) {
+        BaseString::snprintf(buf1, sizeof(buf1), "%s could not allocate memory for parameter %s", 
+	         getBlockName(number()), param_info.m_name);
+      } else {
         BaseString::snprintf(buf1, sizeof(buf1), "%s could not allocate memory for %s", 
 	         getBlockName(number()), type);
-      BaseString::snprintf(buf2, sizeof(buf2), "Requested: %ux%u = %u bytes", 
-	       (Uint32)s, (Uint32)n, (Uint32)size);
+      }
+      BaseString::snprintf(buf2, sizeof(buf2), "Requested: %ux%u = %llu bytes", 
+	       (Uint32)s, (Uint32)n, (Uint64)real_size);
       ERROR_SET(fatal, NDBD_EXIT_MEMALLOC, buf1, buf2);
     }
 

storage/ndb/src/kernel/vm/SimulatedBlock.hpp

@@ -377,7 +377,7 @@ protected:
    * Allocates memory for the datastructures where ndb keeps the data
    *
    */
-  void* allocRecord(const char * type, size_t s, size_t n, bool clear = true);
+  void* allocRecord(const char * type, size_t s, size_t n, bool clear = true, Uint32 paramId = 0);
   
   /**
    * Deallocate record

storage/ndb/src/mgmapi/Makefile.am

 
 noinst_LTLIBRARIES = libmgmapi.la
 
-libmgmapi_la_SOURCES = mgmapi.cpp ndb_logevent.cpp mgmapi_configuration.cpp LocalConfig.cpp ../kernel/error/ndbd_exit_codes.c
+libmgmapi_la_SOURCES = mgmapi.cpp ndb_logevent.cpp mgmapi_configuration.cpp LocalConfig.cpp ../kernel/error/ndbd_exit_codes.c ../mgmsrv/ParamInfo.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/include/mgmapi
 

storage/ndb/src/mgmapi/mgmapi_configuration.cpp

 #include <ndb_types.h>
 #include <mgmapi.h>
 #include "mgmapi_configuration.hpp"
+#include "../mgmsrv/ParamInfo.hpp"
+
+extern const ParamInfo ParamInfoArray[];
+extern const int ParamInfoNum;
 
 ndb_mgm_configuration_iterator::ndb_mgm_configuration_iterator
 (const ndb_mgm_configuration & conf, unsigned type_of_section)
@@ -155,3 +159,37 @@ ndb_mgm_find(ndb_mgm_configuration_iterator* iter,
 	     int param, unsigned search){
   return iter->find(param, search);
 }
+
+/**
+ * Retrieve information about parameter
+ * @param info : in - pointer to structure allocated by caller
+ * @param size : in/out : pointer to int initialized to sizeof(ndb_mgm_param_info)...will be set to bytes set by function on return
+*/
+extern "C"
+int 
+ndb_mgm_get_db_parameter_info(Uint32 paramId, struct ndb_mgm_param_info * info, size_t * size) {
+  if ( paramId == 0 ) {
+      return -1;
+  }
+
+  for (int i = 0; i < ParamInfoNum; i++) {
+    if (paramId == ParamInfoArray[i]._paramId && strcmp(DB_TOKEN, ParamInfoArray[i]._section) == 0) {
+        size_t tmp = 0;
+        if (tmp + sizeof(info->m_id) <= *size)
+        {
+          info->m_id = ParamInfoArray[i]._paramId;
+          tmp += sizeof(info->m_id);
+        }
+
+        if (tmp + sizeof(info->m_name) <= *size)
+        {
+          info->m_name = ParamInfoArray[i]._fname;
+          tmp += sizeof(info->m_name);
+        }
+
+        *size = tmp;
+        return 0;
+    }
+  }
+  return -1;
+}

storage/ndb/src/mgmsrv/ConfigInfo.cpp

@@ -3151,6 +3151,30 @@ checkDbConstraints(InitConfigFileParser::Context & ctx, const char *){
     ctx.m_userProperties.put("NoOfReplicas", replicas);
   }
 
+  /**
+   * In kernel, will calculate the MaxNoOfMeataTables use the following sum:
+   * Uint32 noOfMetaTables = noOfTables + noOfOrderedIndexes + 
+   *                         noOfUniqueHashIndexes + 2
+   * 2 is the number of the SysTables.
+   * So must check that the sum does't exceed the max value of Uint32.
+   */
+  Uint32 noOfTables = 0,
+         noOfOrderedIndexes = 0,
+         noOfUniqueHashIndexes = 0;
+  ctx.m_currentSection->get("MaxNoOfTables", &noOfTables);
+  ctx.m_currentSection->get("MaxNoOfOrderedIndexes", &noOfOrderedIndexes);
+  ctx.m_currentSection->get("MaxNoOfUniqueHashIndexes", &noOfUniqueHashIndexes);
+
+  Uint64 sum= (Uint64)noOfTables + noOfOrderedIndexes + noOfUniqueHashIndexes;
+  
+  if (sum > ((Uint32)~0 - 2)) {
+    ctx.reportError("The sum of MaxNoOfTables, MaxNoOfOrderedIndexes and"
+		    " MaxNoOfUniqueHashIndexes must not exceed %u - [%s]"
+                    " starting at line: %d",
+		    ((Uint32)~0 - 2), ctx.fname, ctx.m_sectionLineno);
+    return false;
+  } 
+
   return true;
 }
 

storage/ndb/src/mgmsrv/ParamInfo.hpp

+#ifndef PARAMINFO_H
+#define PARAMINFO_H
+
+#define DB_TOKEN "DB"
+#define MGM_TOKEN "MGM"
+#define API_TOKEN "API"
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+/**
+ * The Configuration parameter type and status
+ */
+
+enum ParameterType        { CI_BOOL, CI_INT, CI_INT64, CI_STRING, CI_SECTION };
+enum ParameterStatus      { CI_USED,            ///< Active
+		     CI_DEPRICATED,      ///< Can be, but shouldn't
+		     CI_NOTIMPLEMENTED,  ///< Is ignored.
+		     CI_INTERNAL         ///< Not configurable by the user
+};
+
+/**
+ *   Entry for one configuration parameter
+ */
+typedef struct m_ParamInfo {
+  Uint32         _paramId;
+  const char*    _fname;   
+  const char*    _section;
+  const char*    _description;
+  ParameterStatus         _status;
+  bool           _updateable;    
+  ParameterType           _type;          
+  const char*    _default;
+  const char*    _min;
+  const char*    _max;
+}ParamInfo;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

storage/ndb/src/mgmsrv/Services.cpp

@@ -1352,7 +1352,7 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData, NodeId nodeId)
   if (EventLoggerBase::event_lookup(eventType,cat,threshold,severity,textF))
     DBUG_VOID_RETURN;
 
-  char m_text[256];
+  char m_text[512];
   EventLogger::getText(m_text, sizeof(m_text),
 		       textF, theData, nodeId);
 
@@ -1368,6 +1368,15 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData, NodeId nodeId)
     if (ndb_logevent_body[i].index_fn)
       val= (*(ndb_logevent_body[i].index_fn))(val);
     str.appfmt("%s=%d\n",ndb_logevent_body[i].token, val);
+    if(strcmp(ndb_logevent_body[i].token,"error") == 0)
+    {
+      int m_text_len= strlen(m_text);
+      if(sizeof(m_text)-m_text_len-3 > 0)
+      {
+        BaseString::snprintf(m_text+m_text_len, 4 , " - ");
+        ndb_error_string(val, m_text+(m_text_len+3), sizeof(m_text)-m_text_len-3);
+      }
+    }
   }
   
   Vector<NDB_SOCKET_TYPE> copy;

storage/ndb/src/ndbapi/ndberror.c

@@ -774,11 +774,14 @@ const char *ndberror_classification_message(ndberror_classification classificati
   return empty_string;
 }
 
-int ndb_error_string(int err_no, char *str, unsigned int size)
+int ndb_error_string(int err_no, char *str, int size)
 {
   ndberror_struct error;
-  unsigned int len;
+  int len;
 
+  assert(size > 1);
+  if(size <= 1) 
+    return 0;
   error.code = err_no;
   ndberror_update(&error);
 

storage/ndb/tools/restore/consumer.hpp

@@ -36,6 +36,7 @@ public:
   virtual void logEntry(const LogEntry &){}
   virtual void endOfLogEntrys(){}
   virtual bool finalize_table(const TableS &){return true;}
+  virtual bool createSystable(const TableS &){ return true;}
   virtual bool update_apply_status(const RestoreMetaData &metaData){return true;}
   NODE_GROUP_MAP *m_nodegroup_map;
   uint            m_nodegroup_map_len;

storage/ndb/tools/restore/consumer_restore.cpp

@@ -666,6 +666,33 @@ err:
   return result;
 }
 
+bool
+BackupRestore::createSystable(const TableS & tables){
+  const char *tablename = tables.getTableName();
+
+  if( strcmp(tablename, NDB_REP_DB "/def/" NDB_APPLY_TABLE) != 0 &&
+      strcmp(tablename, NDB_REP_DB "/def/" NDB_SCHEMA_TABLE) != 0 )
+  {
+    return true;
+  }
+
+  BaseString tmp(tablename);
+  Vector<BaseString> split;
+  if(tmp.split(split, "/") != 3){
+    err << "Invalid table name format " << tablename << endl;
+    return false;
+  }
+
+  m_ndb->setDatabaseName(split[0].c_str());
+  m_ndb->setSchemaName(split[1].c_str());
+
+  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
+  if( dict->getTable(split[2].c_str()) != NULL ){
+    return true;
+  }
+  return table(tables);
+}
+
 bool
 BackupRestore::table(const TableS & table){
   if (!m_restore && !m_restore_meta)

storage/ndb/tools/restore/consumer_restore.hpp

@@ -73,6 +73,7 @@ public:
   virtual void endOfLogEntrys();
   virtual bool finalize_table(const TableS &);
   virtual bool has_temp_error();
+  virtual bool createSystable(const TableS & table);
   virtual bool update_apply_status(const RestoreMetaData &metaData);
   void connectToMysql();
   bool map_in_frm(char *new_data, const char *data,

storage/ndb/tools/restore/restore_main.cpp

@@ -567,6 +567,15 @@ main(int argc, char** argv)
           err << metaData[i]->getTableName() << " ... Exiting " << endl;
 	  exitHandler(NDBT_FAILED);
 	} 
+    } else {
+      for(Uint32 j= 0; j < g_consumers.size(); j++)
+        if (!g_consumers[j]->createSystable(* metaData[i]))
+        {
+          err << "Restore: Failed to restore system table: ";
+          err << metaData[i]->getTableName() << " ... Exiting " << endl;
+          exitHandler(NDBT_FAILED);
+        }
+
     }
   }
   debug << "Close tables" << endl;