ndb - recommit extra version info to real-51

storage/ndb/include/kernel/GlobalSignalNumbers.h:
  add extra version info to main clone
storage/ndb/include/kernel/NodeInfo.hpp:
  add extra version info to main clone
storage/ndb/include/kernel/signaldata/ApiRegSignalData.hpp:
  add extra version info to main clone
storage/ndb/include/ndb_version.h.in:
  add extra version info to main clone
storage/ndb/src/common/debugger/signaldata/SignalNames.cpp:
  add extra version info to main clone
storage/ndb/src/kernel/blocks/qmgr/Qmgr.hpp:
  add extra version info to main clone
storage/ndb/src/kernel/blocks/qmgr/QmgrInit.cpp:
  add extra version info to main clone
storage/ndb/src/kernel/blocks/qmgr/QmgrMain.cpp:
  add extra version info to main clone
storage/ndb/src/kernel/vm/GlobalData.hpp:
  add extra version info to main clone
storage/ndb/src/kernel/vm/SimulatedBlock.hpp:
  add extra version info to main clone

storage/ndb/include/kernel/GlobalSignalNumbers.h

@@ -183,7 +183,7 @@ extern const GlobalSignalNumber NO_OF_SIGNAL_NAMES;
 #define GSN_CNTR_START_REP              119
 /* 120 not unused */
 #define GSN_ROUTE_ORD                   121
-/* 122 unused */
+#define GSN_NODE_VERSION_REP            122
 /* 123 unused */
 /* 124 unused */
 #define GSN_CHECK_LCP_STOP              125

storage/ndb/include/kernel/NodeInfo.hpp

@@ -90,4 +90,14 @@ operator<<(NdbOut& ndbout, const NodeInfo & info){
   return ndbout;
 }
 
+struct NodeVersionInfo
+{
+  STATIC_CONST( DataLength = 6 );
+  struct 
+  {
+    Uint32 m_min_version;
+    Uint32 m_max_version;
+  } m_type [3]; // Indexed as NodeInfo::Type 
+};
+
 #endif

storage/ndb/include/kernel/signaldata/ApiRegSignalData.hpp

@@ -80,12 +80,13 @@ class ApiRegConf {
   friend class ClusterMgr;
 
 public:
-  STATIC_CONST( SignalLength = 3 + NodeState::DataLength );
+  STATIC_CONST( SignalLength = 4 + NodeState::DataLength );
 private:
   
   Uint32 qmgrRef;
   Uint32 version; // Version of NDB node
   Uint32 apiHeartbeatFrequency;
+  Uint32 minDbVersion;
   NodeState nodeState;
 };
 

storage/ndb/include/ndb_version.h.in

@@ -72,5 +72,7 @@ char ndb_version_string_buf[NDB_VERSION_STRING_BUF_SZ];
 
 #define NDBD_QMGR_SINGLEUSER_VERSION_5 MAKE_VERSION(5,0,25)
 
+#define NDBD_NODE_VERSION_REP MAKE_VERSION(6,1,1)
+
 #endif
  

storage/ndb/src/common/debugger/signaldata/SignalNames.cpp

@@ -637,5 +637,6 @@ const GsnName SignalNames [] = {
   ,{ GSN_DICT_COMMIT_REQ,  "DICT_COMMIT_REQ"}
 
   ,{ GSN_ROUTE_ORD, "ROUTE_ORD" }
+  ,{ GSN_NODE_VERSION_REP, "NODE_VERSION_REP" }
 };
 const unsigned short NO_OF_SIGNAL_NAMES = sizeof(SignalNames)/sizeof(GsnName);

storage/ndb/src/kernel/blocks/qmgr/Qmgr.hpp

@@ -450,6 +450,11 @@ private:
 #ifdef ERROR_INSERT
   Uint32 c_error_insert_extra;
 #endif
+
+  void recompute_version_info(Uint32 type);
+  void recompute_version_info(Uint32 type, Uint32 version);
+  void execNODE_VERSION_REP(Signal* signal);
+  void sendApiVersionRep(Signal* signal, NodeRecPtr nodePtr);
 };
 
 #endif

storage/ndb/src/kernel/blocks/qmgr/QmgrInit.cpp

@@ -37,6 +37,13 @@ void Qmgr::initData()
   setHbApiDelay(hbDBAPI);
   c_connectedNodes.set(getOwnNodeId());
   c_stopReq.senderRef = 0;
+
+  /**
+   * Check sanity for NodeVersion
+   */
+  ndbrequire((Uint32)NodeInfo::DB == 0);
+  ndbrequire((Uint32)NodeInfo::API == 1);
+  ndbrequire((Uint32)NodeInfo::MGM == 2); 
 }//Qmgr::initData()
 
 void Qmgr::initRecords() 
@@ -107,6 +114,7 @@ Qmgr::Qmgr(Block_context& ctx)
 
   addRecSignal(GSN_DIH_RESTARTREF, &Qmgr::execDIH_RESTARTREF);
   addRecSignal(GSN_DIH_RESTARTCONF, &Qmgr::execDIH_RESTARTCONF);
+  addRecSignal(GSN_NODE_VERSION_REP, &Qmgr::execNODE_VERSION_REP);
   
   initData();
 }//Qmgr::Qmgr()

storage/ndb/src/kernel/blocks/qmgr/QmgrMain.cpp

@@ -260,6 +260,9 @@ void Qmgr::execSTTOR(Signal* signal)
   case 1:
     initData(signal);
     startphase1(signal);
+    recompute_version_info(NodeInfo::DB);
+    recompute_version_info(NodeInfo::API);
+    recompute_version_info(NodeInfo::MGM);
     return;
   case 7:
     cactivateApiCheck = 1;
@@ -765,6 +768,7 @@ void Qmgr::execCM_REGREQ(Signal* signal)
    */
   UintR TdynId = ++c_maxDynamicId;
   setNodeInfo(addNodePtr.i).m_version = startingVersion;
+  recompute_version_info(NodeInfo::DB, startingVersion);
   addNodePtr.p->ndynamicId = TdynId;
   
   /**
@@ -1503,7 +1507,8 @@ void Qmgr::execCM_NODEINFOCONF(Signal* signal)
   replyNodePtr.p->ndynamicId = dynamicId;
   replyNodePtr.p->blockRef = signal->getSendersBlockRef();
   setNodeInfo(replyNodePtr.i).m_version = version;
-
+  recompute_version_info(NodeInfo::DB, version);
+  
   if(!c_start.m_nodes.done()){
     jam();
     return;
@@ -1602,6 +1607,7 @@ Qmgr::cmAddPrepare(Signal* signal, NodeRecPtr nodePtr, const NodeRec * self){
   }
   
   sendCmAckAdd(signal, nodePtr.i, CmAdd::Prepare);
+  sendApiVersionRep(signal, nodePtr);
 
   /* President have prepared us */
   CmNodeInfoConf * conf = (CmNodeInfoConf*)signal->getDataPtrSend();
@@ -1613,6 +1619,29 @@ Qmgr::cmAddPrepare(Signal* signal, NodeRecPtr nodePtr, const NodeRec * self){
   DEBUG_START(GSN_CM_NODEINFOCONF, refToNode(nodePtr.p->blockRef), "");
 }
 
+void
+Qmgr::sendApiVersionRep(Signal* signal, NodeRecPtr nodePtr)
+{
+  if (getNodeInfo(nodePtr.i).m_version >= NDBD_NODE_VERSION_REP)
+  {
+    jam();
+    Uint32 ref = calcQmgrBlockRef(nodePtr.i);
+    for(Uint32 i = 1; i<MAX_NODES; i++)
+    {
+      jam();
+      Uint32 version = getNodeInfo(i).m_version;
+      Uint32 type = getNodeInfo(i).m_type;
+      if (type != NodeInfo::DB && version)
+      {
+	jam();
+	signal->theData[0] = i;
+	signal->theData[1] = version;
+	sendSignal(ref, GSN_NODE_VERSION_REP, signal, 2, JBB);
+      }
+    }
+  }
+}
+
 void
 Qmgr::sendCmAckAdd(Signal * signal, Uint32 nodeId, CmAdd::RequestType type){
   
@@ -2401,7 +2430,9 @@ void Qmgr::sendApiFailReq(Signal* signal, Uint16 failedNodeNo)
    * SECONDS. 
    *-------------------------------------------------------------------------*/
   setNodeInfo(failedNodePtr.i).m_heartbeat_cnt= 0;
-
+  setNodeInfo(failedNodePtr.i).m_version = 0;
+  recompute_version_info(getNodeInfo(failedNodePtr.i).m_type);
+  
   CloseComReqConf * const closeCom = (CloseComReqConf *)&signal->theData[0];
 
   closeCom->xxxBlockRef = reference();
@@ -2707,7 +2738,6 @@ void Qmgr::execAPI_REGREQ(Signal* signal)
   }
 
   setNodeInfo(apiNodePtr.i).m_version = version;
-
   setNodeInfo(apiNodePtr.i).m_heartbeat_cnt= 0;
 
   ApiRegConf * const apiRegConf = (ApiRegConf *)&signal->theData[0];
@@ -2728,8 +2758,9 @@ void Qmgr::execAPI_REGREQ(Signal* signal)
       apiRegConf->nodeState.dynamicId = -dynamicId;
     }
   }
+  NodeVersionInfo info = getNodeVersionInfo();
+  apiRegConf->minDbVersion = info.m_type[NodeInfo::DB].m_min_version;
   apiRegConf->nodeState.m_connected_nodes.assign(c_connectedNodes);
-
   sendSignal(ref, GSN_API_REGCONF, signal, ApiRegConf::SignalLength, JBB);
 
   if (apiNodePtr.p->phase == ZAPI_INACTIVE &&
@@ -2748,6 +2779,33 @@ void Qmgr::execAPI_REGREQ(Signal* signal)
     signal->theData[0] = apiNodePtr.i;
     sendSignal(CMVMI_REF, GSN_ENABLE_COMORD, signal, 1, JBA);
     
+    recompute_version_info(type, version);
+    
+    if (info.m_type[NodeInfo::DB].m_min_version >= NDBD_NODE_VERSION_REP)
+    {
+      jam();
+      NodeReceiverGroup rg(QMGR, c_clusterNodes);
+      rg.m_nodes.clear(getOwnNodeId());
+      signal->theData[0] = apiNodePtr.i;
+      signal->theData[1] = version;
+      sendSignal(rg, GSN_NODE_VERSION_REP, signal, 2, JBB);
+    }
+    else
+    {
+      Uint32 i = 0;
+      while((i = c_clusterNodes.find(i + 1)) != NdbNodeBitmask::NotFound)
+      {
+	jam();
+	if (i == getOwnNodeId())
+	  continue;
+	if (getNodeInfo(i).m_version >= NDBD_NODE_VERSION_REP)
+	{
+	  jam();
+	  sendSignal(calcQmgrBlockRef(i), GSN_NODE_VERSION_REP, signal, 2,JBB);
+	}
+      }
+    }
+    
     signal->theData[0] = apiNodePtr.i;
     EXECUTE_DIRECT(NDBCNTR, GSN_API_START_REP, signal, 1);
   }
@@ -2783,6 +2841,76 @@ Qmgr::execAPI_VERSION_REQ(Signal * signal) {
 	     ApiVersionConf::SignalLength, JBB);
 }
 
+void
+Qmgr::execNODE_VERSION_REP(Signal* signal)
+{
+  jamEntry();
+  Uint32 nodeId = signal->theData[0];
+  Uint32 version = signal->theData[1];
+
+  if (nodeId < MAX_NODES)
+  {
+    jam();
+    Uint32 type = getNodeInfo(nodeId).m_type;
+    setNodeInfo(nodeId).m_version = version;
+    recompute_version_info(type, version);
+  }
+}
+ 
+void
+Qmgr::recompute_version_info(Uint32 type, Uint32 version)
+{
+  NodeVersionInfo& info = setNodeVersionInfo();
+  switch(type){
+  case NodeInfo::DB:
+  case NodeInfo::API:
+  case NodeInfo::MGM:
+    break;
+  default:
+    return;
+  }
+  
+  if (info.m_type[type].m_min_version == 0 ||
+      version < info.m_type[type].m_min_version)
+    info.m_type[type].m_min_version = version;
+  if (version > info.m_type[type].m_max_version)
+    info.m_type[type].m_max_version = version;
+}
+
+void
+Qmgr::recompute_version_info(Uint32 type)
+{
+  switch(type){
+  case NodeInfo::DB:
+  case NodeInfo::API:
+  case NodeInfo::MGM:
+    break;
+  default:
+    return;
+  }
+  
+  Uint32 min = ~0, max = 0;
+  Uint32 cnt = type == NodeInfo::DB ? MAX_NDB_NODES : MAX_NODES;
+  for (Uint32 i = 1; i<cnt; i++)
+  {
+    if (getNodeInfo(i).m_type == type)
+    {
+      Uint32 version = getNodeInfo(i).m_version;
+      
+      if (version)
+      {
+	if (version < min)
+	  min = version;
+	if (version > max)
+	  max = version;
+      }
+    }
+  }
+  
+  NodeVersionInfo& info = setNodeVersionInfo();
+  info.m_type[type].m_min_version = min == ~(Uint32)0 ? 0 : min;
+  info.m_type[type].m_max_version = max;
+}
 
 #if 0
 bool

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

@@ -36,6 +36,7 @@ enum  restartStates {initial_state,
 
 struct GlobalData {
   Uint32     m_restart_seq;           // 
+  NodeVersionInfo m_versionInfo;
   NodeInfo   m_nodeInfo[MAX_NODES];
   Signal     VMSignals[1];            // Owned by FastScheduler::
   

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

@@ -403,6 +403,9 @@ protected:
   const NodeInfo & getNodeInfo(NodeId nodeId) const;
   NodeInfo & setNodeInfo(NodeId);
 
+  const NodeVersionInfo& getNodeVersionInfo() const;
+  NodeVersionInfo& setNodeVersionInfo();
+  
   /**********************
    * Xfrm stuff
    */
@@ -709,6 +712,18 @@ SimulatedBlock::getNodeInfo(NodeId nodeId) const {
   return globalData.m_nodeInfo[nodeId];
 }
 
+inline
+const NodeVersionInfo &
+SimulatedBlock::getNodeVersionInfo() const {
+  return globalData.m_versionInfo;
+}
+
+inline
+NodeVersionInfo &
+SimulatedBlock::setNodeVersionInfo() {
+  return globalData.m_versionInfo;
+}
+
 inline
 void
 SimulatedBlock::EXECUTE_DIRECT(Uint32 block,