Bug #25275 SINGLE USER MODE prevents ALTER on non-ndb tables for other mysqld nodes

ndb/include/kernel/signaldata/DictTabInfo.hpp

@@ -126,6 +126,8 @@ public:
     MinRowsLow         = 143,
     MinRowsHigh        = 144,
 
+    SingleUserMode     = 152,
+
     TableEnd           = 999,
     
     AttributeName          = 1000, // String, Mandatory
@@ -273,6 +275,7 @@ public:
     Uint32 MaxRowsHigh;
     Uint32 MinRowsLow;
     Uint32 MinRowsHigh;
+    Uint32 SingleUserMode;
 
     void init();
   };

ndb/include/ndb_constants.h

@@ -68,4 +68,11 @@
 
 #define NDB_TYPE_MAX                    31
  
+/*
+ * Table single user mode
+ */
+#define NDB_SUM_LOCKED       0
+#define NDB_SUM_READONLY     1
+#define NDB_SUM_READ_WRITE   2
+
 #endif

ndb/include/ndbapi/NdbDictionary.hpp

@@ -497,6 +497,15 @@ public:
    */
   class Table : public Object {
   public:
+    /*
+     * Single user mode specifies access rights to table during single user mode
+     */
+    enum SingleUserMode {
+      SingleUserModeLocked    = NDB_SUM_LOCKED,
+      SingleUserModeReadOnly  = NDB_SUM_READONLY,
+      SingleUserModeReadWrite = NDB_SUM_READ_WRITE
+    };
+
     /** 
      * @name General
      * @{
@@ -735,6 +744,13 @@ public:
     void setMinRows(Uint64 minRows);
     Uint64 getMinRows() const;
 
+    /**
+     * Set/Get SingleUserMode
+     */
+    void setSingleUserMode(enum SingleUserMode);
+    enum SingleUserMode getSingleUserMode() const;
+
+
     /** @} *******************************************************************/
 
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL

ndb/src/common/debugger/signaldata/DictTabInfo.cpp

@@ -51,6 +51,7 @@ DictTabInfo::TableMapping[] = {
   DTIMAP(Table, MaxRowsHigh, MaxRowsHigh),
   DTIMAP(Table, MinRowsLow, MinRowsLow),
   DTIMAP(Table, MinRowsHigh, MinRowsHigh),
+  DTIMAP(Table, SingleUserMode, SingleUserMode),
   DTIBREAK(AttributeName)
 };
 
@@ -131,6 +132,8 @@ DictTabInfo::Table::init(){
   MaxRowsHigh = 0;
   MinRowsLow = 0;
   MinRowsHigh = 0;
+
+  SingleUserMode = 0;
 }
 
 void

ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -289,7 +289,7 @@ Dbdict::packTableIntoPagesImpl(SimpleProperties::Writer & w,
   w.add(DictTabInfo::MaxRowsHigh, tablePtr.p->maxRowsHigh);
   w.add(DictTabInfo::MinRowsLow, tablePtr.p->minRowsLow);
   w.add(DictTabInfo::MinRowsHigh, tablePtr.p->minRowsHigh);
-  
+  w.add(DictTabInfo::SingleUserMode, tablePtr.p->singleUserMode);
   if(!signal)
   {
     w.add(DictTabInfo::FragmentCount, tablePtr.p->fragmentCount);
@@ -1500,6 +1500,7 @@ void Dbdict::initialiseTableRecord(TableRecordPtr tablePtr)
   tablePtr.p->maxRowsHigh = 0;
   tablePtr.p->minRowsLow = 0;
   tablePtr.p->minRowsHigh = 0;
+  tablePtr.p->singleUserMode = 0;
   tablePtr.p->storedTable = true;
   tablePtr.p->tableType = DictTabInfo::UserTable;
   tablePtr.p->primaryTableId = RNIL;
@@ -4718,8 +4719,9 @@ Dbdict::execTAB_COMMITCONF(Signal* signal){
     signal->theData[4] = (Uint32)tabPtr.p->tableType;
     signal->theData[5] = createTabPtr.p->key;
     signal->theData[6] = (Uint32)tabPtr.p->noOfPrimkey;
+    signal->theData[7] = (Uint32)tabPtr.p->singleUserMode;
 
-    sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB);
+    sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 8, JBB);
     return;
   }
   
@@ -5084,6 +5086,7 @@ void Dbdict::handleTabInfoInit(SimpleProperties::Reader & it,
   tablePtr.p->maxRowsHigh = tableDesc.MaxRowsHigh;
   tablePtr.p->minRowsLow = tableDesc.MinRowsLow;
   tablePtr.p->minRowsHigh = tableDesc.MinRowsHigh;
+  tablePtr.p->singleUserMode = tableDesc.SingleUserMode;
 
   Uint64 maxRows =
     (((Uint64)tablePtr.p->maxRowsHigh) << 32) + tablePtr.p->maxRowsLow;

ndb/src/kernel/blocks/dbdict/Dbdict.hpp

@@ -237,6 +237,11 @@ public:
     char frmData[MAX_FRM_DATA_SIZE];
 
     Uint32 fragmentCount;
+
+    /*
+     * Access rights to table during single user mode
+     */
+    Uint8 singleUserMode;
   };
 
   typedef Ptr<TableRecord> TableRecordPtr;

ndb/src/kernel/blocks/dbtc/Dbtc.hpp

@@ -702,6 +702,7 @@ public:
     Uint8 tckeyrec; // Ändrad från R
     Uint8 tcindxrec;
     Uint8 apiFailState; // Ändrad från R
+    Uint8 singleUserMode;
     ReturnSignal returnsignal;
     Uint8 timeOutCounter;
     
@@ -957,17 +958,28 @@ public:
   /********************************************************/
   struct TableRecord {
     Uint32 currentSchemaVersion;
-    Uint8 enabled;
+    Uint16 m_flags;
-    Uint8 dropping;
     Uint8 tableType;
-    Uint8 storedTable;
+    Uint8 singleUserMode;
+
+    enum {
+      TR_ENABLED      = 1 << 0,
+      TR_DROPPING     = 1 << 1,
+      TR_STORED_TABLE = 1 << 2
+    };
+    Uint8 get_enabled()     const { return (m_flags & TR_ENABLED)      != 0; }
+    Uint8 get_dropping()    const { return (m_flags & TR_DROPPING)     != 0; }
+    Uint8 get_storedTable() const { return (m_flags & TR_STORED_TABLE) != 0; }
+    void set_enabled(Uint8 f)     { f ? m_flags |= (Uint16)TR_ENABLED      : m_flags &= ~(Uint16)TR_ENABLED; }
+    void set_dropping(Uint8 f)    { f ? m_flags |= (Uint16)TR_DROPPING     : m_flags &= ~(Uint16)TR_DROPPING; }
+    void set_storedTable(Uint8 f) { f ? m_flags |= (Uint16)TR_STORED_TABLE : m_flags &= ~(Uint16)TR_STORED_TABLE; }
 
     Uint8 noOfKeyAttr;
     Uint8 hasCharAttr;
     Uint8 noOfDistrKeys;
     
     bool checkTable(Uint32 schemaVersion) const {
-      return enabled && !dropping && 
+      return get_enabled() && !get_dropping() && 
 	(table_version_major(schemaVersion) == table_version_major(currentSchemaVersion));
     }
 
@@ -1835,10 +1847,10 @@ private:
 			Uint32 transid2);
   void removeMarkerForFailedAPI(Signal* signal, Uint32 nodeId, Uint32 bucket);
 
-  bool getAllowStartTransaction(Uint32 nodeId) const {
+  bool getAllowStartTransaction(Uint32 nodeId, Uint32 table_single_user_mode) const {
     if (unlikely(getNodeState().getSingleUserMode()))
     {
-      if (getNodeState().getSingleUserApi() == nodeId)
+      if (getNodeState().getSingleUserApi() == nodeId || table_single_user_mode)
         return true;
       else
         return false;

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -327,19 +327,21 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
   tabptr.i = signal->theData[0];
   ptrCheckGuard(tabptr, ctabrecFilesize, tableRecord);
   tabptr.p->currentSchemaVersion = signal->theData[1];
-  tabptr.p->storedTable = (bool)signal->theData[2];
+  tabptr.p->m_flags = 0;
+  tabptr.p->set_storedTable((bool)signal->theData[2]);
   BlockReference retRef = signal->theData[3];
   tabptr.p->tableType = (Uint8)signal->theData[4];
   BlockReference retPtr = signal->theData[5];
   Uint32 noOfKeyAttr = signal->theData[6];
+  tabptr.p->singleUserMode = (Uint8)signal->theData[7];
   ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
 
   const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tabptr.i);
   ndbrequire(noOfKeyAttr == desc->noOfKeyAttr);
 
-  ndbrequire(tabptr.p->enabled == false);
+  ndbrequire(tabptr.p->get_enabled() == false);
-  tabptr.p->enabled = true;
+  tabptr.p->set_enabled(true);
-  tabptr.p->dropping = false;
+  tabptr.p->set_dropping(false);
   tabptr.p->noOfKeyAttr = desc->noOfKeyAttr;
   tabptr.p->hasCharAttr = desc->hasCharAttr;
   tabptr.p->noOfDistrKeys = desc->noOfDistrKeys;
@@ -363,7 +365,7 @@ Dbtc::execPREP_DROP_TAB_REQ(Signal* signal)
   Uint32 senderRef = req->senderRef;
   Uint32 senderData = req->senderData;
   
-  if(!tabPtr.p->enabled){
+  if(!tabPtr.p->get_enabled()){
     jam();
     PrepDropTabRef* ref = (PrepDropTabRef*)signal->getDataPtrSend();
     ref->senderRef = reference();
@@ -375,7 +377,7 @@ Dbtc::execPREP_DROP_TAB_REQ(Signal* signal)
     return;
   }
 
-  if(tabPtr.p->dropping){
+  if(tabPtr.p->get_dropping()){
     jam();
     PrepDropTabRef* ref = (PrepDropTabRef*)signal->getDataPtrSend();
     ref->senderRef = reference();
@@ -387,7 +389,7 @@ Dbtc::execPREP_DROP_TAB_REQ(Signal* signal)
     return;
   }
   
-  tabPtr.p->dropping = true;
+  tabPtr.p->set_dropping(true);
   tabPtr.p->dropTable.senderRef = senderRef;
   tabPtr.p->dropTable.senderData = senderData;
 
@@ -423,7 +425,7 @@ Dbtc::execWAIT_DROP_TAB_CONF(Signal* signal)
   tabPtr.i = conf->tableId;
   ptrCheckGuard(tabPtr, ctabrecFilesize, tableRecord);
   
-  ndbrequire(tabPtr.p->dropping == true);
+  ndbrequire(tabPtr.p->get_dropping() == true);
   Uint32 nodeId = refToNode(conf->senderRef);
   tabPtr.p->dropTable.waitDropTabCount.clearWaitingFor(nodeId);
   
@@ -453,7 +455,7 @@ Dbtc::execWAIT_DROP_TAB_REF(Signal* signal)
   tabPtr.i = ref->tableId;
   ptrCheckGuard(tabPtr, ctabrecFilesize, tableRecord);
   
-  ndbrequire(tabPtr.p->dropping == true);
+  ndbrequire(tabPtr.p->get_dropping() == true);
   Uint32 nodeId = refToNode(ref->senderRef);
   tabPtr.p->dropTable.waitDropTabCount.clearWaitingFor(nodeId);
   
@@ -490,7 +492,7 @@ Dbtc::checkWaitDropTabFailedLqh(Signal* signal, Uint32 nodeId, Uint32 tableId)
   for(Uint32 i = 0; i<RT_BREAK && tabPtr.i < ctabrecFilesize; i++, tabPtr.i++){
     jam();
     ptrAss(tabPtr, tableRecord);
-    if(tabPtr.p->enabled && tabPtr.p->dropping){
+    if(tabPtr.p->get_enabled() && tabPtr.p->get_dropping()){
       if(tabPtr.p->dropTable.waitDropTabCount.isWaitingFor(nodeId)){
         jam();
 	conf->senderRef = calcLqhBlockRef(nodeId);
@@ -531,7 +533,7 @@ Dbtc::execDROP_TAB_REQ(Signal* signal)
   Uint32 senderData = req->senderData;
   DropTabReq::RequestType rt = (DropTabReq::RequestType)req->requestType;
   
-  if(!tabPtr.p->enabled && rt == DropTabReq::OnlineDropTab){
+  if(!tabPtr.p->get_enabled() && rt == DropTabReq::OnlineDropTab){
     jam();
     DropTabRef* ref = (DropTabRef*)signal->getDataPtrSend();
     ref->senderRef = reference();
@@ -543,7 +545,7 @@ Dbtc::execDROP_TAB_REQ(Signal* signal)
     return;
   }
 
-  if(!tabPtr.p->dropping && rt == DropTabReq::OnlineDropTab){
+  if(!tabPtr.p->get_dropping() && rt == DropTabReq::OnlineDropTab){
     jam();
     DropTabRef* ref = (DropTabRef*)signal->getDataPtrSend();
     ref->senderRef = reference();
@@ -555,8 +557,8 @@ Dbtc::execDROP_TAB_REQ(Signal* signal)
     return;
   }
   
-  tabPtr.p->enabled = false;
+  tabPtr.p->set_enabled(false);
-  tabPtr.p->dropping = false;
+  tabPtr.p->set_dropping(false);
   
   DropTabConf * conf = (DropTabConf*)signal->getDataPtrSend();
   conf->tableId = tabPtr.i;
@@ -1214,8 +1216,7 @@ void Dbtc::execTCSEIZEREQ(Signal* signal)
 	      break;
 	    case NodeState::SL_STOPPING_1:
 	    case NodeState::SL_STOPPING_2:
-              if (getNodeState().getSingleUserMode() &&
+              if (getNodeState().getSingleUserMode())
-                  getNodeState().getSingleUserApi() == senderNodeId)
                 break;
 	    case NodeState::SL_STOPPING_3:
 	    case NodeState::SL_STOPPING_4:
@@ -1225,9 +1226,6 @@ void Dbtc::execTCSEIZEREQ(Signal* signal)
 		errCode = ZNODE_SHUTDOWN_IN_PROGRESS;
 	      break;
 	    case NodeState::SL_SINGLEUSER:
-              if (getNodeState().getSingleUserApi() == senderNodeId)
-                break;
-	      errCode = ZCLUSTER_IN_SINGLEUSER_MODE;
 	      break;
 	    default:
 	      errCode = ZWRONG_STATE;
@@ -2397,6 +2395,7 @@ void Dbtc::initApiConnectRec(Signal* signal,
   regApiPtr->buddyPtr = RNIL;
   regApiPtr->currSavePointId = 0;
   regApiPtr->m_transaction_nodes.clear();
+  regApiPtr->singleUserMode = 0;
   // Trigger data
   releaseFiredTriggerData(&regApiPtr->theFiredTriggers),
   // Index data
@@ -2550,9 +2549,12 @@ void Dbtc::execTCKEYREQ(Signal* signal)
   bool isIndexOpReturn = regApiPtr->indexOpReturn;
   regApiPtr->isIndexOp = false; // Reset marker
   regApiPtr->m_exec_flag |= TexecFlag;
+  TableRecordPtr localTabptr;
+  localTabptr.i = TtabIndex;
+  localTabptr.p = &tableRecord[TtabIndex];
   switch (regApiPtr->apiConnectstate) {
   case CS_CONNECTED:{
-    if (TstartFlag == 1 && getAllowStartTransaction(sendersNodeId) == true){
+    if (TstartFlag == 1 && getAllowStartTransaction(sendersNodeId, localTabptr.p->singleUserMode) == true){
       //---------------------------------------------------------------------
       // Initialise API connect record if transaction is started.
       //---------------------------------------------------------------------
@@ -2560,7 +2562,7 @@ void Dbtc::execTCKEYREQ(Signal* signal)
       initApiConnectRec(signal, regApiPtr);
       regApiPtr->m_exec_flag = TexecFlag;
     } else {
-      if(getAllowStartTransaction(sendersNodeId) == true){
+      if(getAllowStartTransaction(sendersNodeId, localTabptr.p->singleUserMode) == true){
 	/*------------------------------------------------------------------
 	 * WE EXPECTED A START TRANSACTION. SINCE NO OPERATIONS HAVE BEEN 
 	 * RECEIVED WE INDICATE THIS BY SETTING FIRST_TC_CONNECT TO RNIL TO 
@@ -2587,6 +2589,13 @@ void Dbtc::execTCKEYREQ(Signal* signal)
        *  the state will be CS_STARTED
        */
       jam();
+      if (unlikely(getNodeState().getSingleUserMode()) &&
+          getNodeState().getSingleUserApi() != sendersNodeId &&
+          !localTabptr.p->singleUserMode)
+      {
+	TCKEY_abort(signal, TexecFlag ? 60 : 57);
+        return;
+      }
       initApiConnectRec(signal, regApiPtr);
       regApiPtr->m_exec_flag = TexecFlag;
     } else { 
@@ -2607,6 +2616,10 @@ void Dbtc::execTCKEYREQ(Signal* signal)
   case CS_ABORTING:
     if (regApiPtr->abortState == AS_IDLE) {
       if (TstartFlag == 1) {
+        if(getAllowStartTransaction(sendersNodeId, localTabptr.p->singleUserMode) == false){
+          TCKEY_abort(signal, TexecFlag ? 60 : 57);
+          return;
+        }
 	//--------------------------------------------------------------------
 	// Previous transaction had been aborted and the abort was completed. 
 	// It is then OK to start a new transaction again.
@@ -2670,9 +2683,6 @@ void Dbtc::execTCKEYREQ(Signal* signal)
     return;
   }//switch
   
-  TableRecordPtr localTabptr;
-  localTabptr.i = TtabIndex;
-  localTabptr.p = &tableRecord[TtabIndex];
   if (localTabptr.p->checkTable(tcKeyReq->tableSchemaVersion)) {
     ;
   } else {
@@ -2731,6 +2741,8 @@ void Dbtc::execTCKEYREQ(Signal* signal)
   regTcPtr->savePointId = regApiPtr->currSavePointId;
   regApiPtr->executingIndexOp = RNIL;
 
+  regApiPtr->singleUserMode |= 1 << localTabptr.p->singleUserMode;
+
   if (TcKeyReq::getExecutingTrigger(Treqinfo)) {
     // Save the TcOperationPtr for fireing operation
     regTcPtr->triggeringOperation = TsenderData;
@@ -2862,7 +2874,7 @@ void Dbtc::execTCKEYREQ(Signal* signal)
      *   THIS VARIABLE CONTROLS THE INTERVAL BETWEEN LCP'S AND 
      *   TEMP TABLES DON'T PARTICIPATE.
      * -------------------------------------------------------------------- */
-    if (localTabptr.p->storedTable) {
+    if (localTabptr.p->get_storedTable()) {
       coperationsize = ((Toperationsize + TattrLen) + TkeyLength) + 17;
     }
     c_counters.cwriteCount = TwriteCount + 1;
@@ -4721,6 +4733,7 @@ void Dbtc::copyApi(Signal* signal)
   regApiPtr->lqhkeyconfrec = Tlqhkeyconfrec;
   regApiPtr->commitAckMarker = TcommitAckMarker;
   regApiPtr->m_transaction_nodes = Tnodes;
+  regApiPtr->singleUserMode = 0;
 
   gcpPtr.i = TgcpPointer;
   ptrCheckGuard(gcpPtr, TgcpFilesize, localGcpRecord);
@@ -4732,6 +4745,7 @@ void Dbtc::copyApi(Signal* signal)
   regTmpApiPtr->firstTcConnect = RNIL;
   regTmpApiPtr->lastTcConnect = RNIL;
   regTmpApiPtr->m_transaction_nodes.clear();
+  regTmpApiPtr->singleUserMode = 0;
   releaseAllSeizedIndexOperations(regTmpApiPtr);
 }//Dbtc::copyApi()
 
@@ -6237,8 +6251,9 @@ void Dbtc::timeOutLoopStartLab(Signal* signal, Uint32 api_con_ptr)
       {
         apiConnectptr.i = api_con_ptr;
         ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
-        if (getNodeState().getSingleUserApi() ==
+        if ((getNodeState().getSingleUserApi() ==
-            refToNode(apiConnectptr.p->ndbapiBlockref))
+             refToNode(apiConnectptr.p->ndbapiBlockref)) ||
+            !(apiConnectptr.p->singleUserMode & (1 << NDB_SUM_LOCKED)))
         {
           // api allowed during single user, use original timeout
           time_out_value=
@@ -8181,6 +8196,7 @@ void Dbtc::initApiConnectFail(Signal* signal)
   apiConnectptr.p->ndbapiConnect = 0;
   apiConnectptr.p->buddyPtr = RNIL;
   apiConnectptr.p->m_transaction_nodes.clear();
+  apiConnectptr.p->singleUserMode = 0;
   setApiConTimer(apiConnectptr.i, 0, __LINE__);
   switch(ttransStatus){
   case LqhTransConf::Committed:
@@ -10076,6 +10092,7 @@ void Dbtc::initApiConnect(Signal* signal)
     apiConnectptr.p->buddyPtr = RNIL;
     apiConnectptr.p->currSavePointId = 0;
     apiConnectptr.p->m_transaction_nodes.clear();
+    apiConnectptr.p->singleUserMode = 0;
   }//for
   apiConnectptr.i = tiacTmp - 1;
   ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
@@ -10104,6 +10121,7 @@ void Dbtc::initApiConnect(Signal* signal)
       apiConnectptr.p->buddyPtr = RNIL;
       apiConnectptr.p->currSavePointId = 0;
       apiConnectptr.p->m_transaction_nodes.clear();
+      apiConnectptr.p->singleUserMode = 0;
     }//for
   apiConnectptr.i = (2 * tiacTmp) - 1;
   ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
@@ -10132,6 +10150,7 @@ void Dbtc::initApiConnect(Signal* signal)
     apiConnectptr.p->buddyPtr = RNIL;
     apiConnectptr.p->currSavePointId = 0;
     apiConnectptr.p->m_transaction_nodes.clear();
+    apiConnectptr.p->singleUserMode = 0;
   }//for
   apiConnectptr.i = (3 * tiacTmp) - 1;
   ptrCheckGuard(apiConnectptr, capiConnectFilesize, apiConnectRecord);
@@ -10316,10 +10335,11 @@ void Dbtc::initTable(Signal* signal)
     refresh_watch_dog();
     ptrAss(tabptr, tableRecord);
     tabptr.p->currentSchemaVersion = 0;
-    tabptr.p->storedTable = true;
+    tabptr.p->m_flags = 0;
+    tabptr.p->set_storedTable(true);
     tabptr.p->tableType = 0;
-    tabptr.p->enabled = false;
+    tabptr.p->set_enabled(false);
-    tabptr.p->dropping = false;
+    tabptr.p->set_dropping(false);
     tabptr.p->noOfKeyAttr = 0;
     tabptr.p->hasCharAttr = 0;
     tabptr.p->noOfDistrKeys = 0;
@@ -10452,6 +10472,7 @@ void Dbtc::releaseAbortResources(Signal* signal)
   apiConnectptr.p->firstTcConnect = RNIL;
   apiConnectptr.p->lastTcConnect = RNIL;
   apiConnectptr.p->m_transaction_nodes.clear();
+  apiConnectptr.p->singleUserMode = 0;
 
   // MASV let state be CS_ABORTING until all 
   // signals in the "air" have been received. Reset to CS_CONNECTED
@@ -11130,7 +11151,7 @@ void Dbtc::execABORT_ALL_REQ(Signal* signal)
   const Uint32 senderData = req->senderData;
   const BlockReference senderRef = req->senderRef;
   
-  if(getAllowStartTransaction(refToNode(senderRef)) == true && !getNodeState().getSingleUserMode()){
+  if(getAllowStartTransaction(refToNode(senderRef), 0) == true && !getNodeState().getSingleUserMode()){
     jam();
 
     ref->senderData = senderData;
@@ -13366,9 +13387,9 @@ void Dbtc::deleteFromIndexTable(Signal* signal,
 
 Uint32 
 Dbtc::TableRecord::getErrorCode(Uint32 schemaVersion) const {
-  if(!enabled)
+  if(!get_enabled())
     return ZNO_SUCH_TABLE;
-  if(dropping)
+  if(get_dropping())
     return ZDROP_TABLE_IN_PROGRESS;
   if(table_version_major(schemaVersion) != table_version_major(currentSchemaVersion))
     return ZWRONG_SCHEMA_VERSION_ERROR;

ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp

@@ -1645,6 +1645,7 @@ void Ndbcntr::createSystableLab(Signal* signal, unsigned index)
   //w.add(DictTabInfo::NoOfVariable, (Uint32)0);
   //w.add(DictTabInfo::KeyLength, 1);
   w.add(DictTabInfo::TableTypeVal, (Uint32)table.tableType);
+  w.add(DictTabInfo::SingleUserMode, (Uint32)NDB_SUM_READ_WRITE);
 
   for (unsigned i = 0; i < table.columnCount; i++) {
     const SysColumn& column = table.columnList[i];

ndb/src/ndbapi/NdbDictionary.cpp

@@ -430,6 +430,18 @@ NdbDictionary::Table::getFrmLength() const {
   return m_impl.m_frm.length();
 }
 
+enum NdbDictionary::Table::SingleUserMode
+NdbDictionary::Table::getSingleUserMode() const
+{
+  return (enum SingleUserMode)m_impl.m_single_user_mode;
+}
+
+void
+NdbDictionary::Table::setSingleUserMode(enum NdbDictionary::Table::SingleUserMode mode)
+{
+  m_impl.m_single_user_mode = (Uint8)mode;
+}
+
 void
 NdbDictionary::Table::setFrm(const void* data, Uint32 len){
   m_impl.m_frm.assign(data, len);

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -318,6 +318,7 @@ NdbTableImpl::init(){
   m_replicaCount= 0;
   m_min_rows = 0;
   m_max_rows = 0;
+  m_single_user_mode = 0;
 }
 
 bool
@@ -378,6 +379,14 @@ NdbTableImpl::equal(const NdbTableImpl& obj) const
     DBUG_RETURN(false);
   }
   
+  if(m_single_user_mode != obj.m_single_user_mode)
+  {
+    DBUG_PRINT("info",("m_single_user_mode %d != %d",
+                       (int32)m_single_user_mode,
+                       (int32)obj.m_single_user_mode));
+    DBUG_RETURN(false);
+  }
+
    DBUG_RETURN(true);
 }
 
@@ -403,6 +412,7 @@ NdbTableImpl::assign(const NdbTableImpl& org)
   m_kvalue = org.m_kvalue;
   m_minLoadFactor = org.m_minLoadFactor;
   m_maxLoadFactor = org.m_maxLoadFactor;
+  m_single_user_mode = org.m_single_user_mode;
 
   if (m_index != 0)
     delete m_index;
@@ -1191,6 +1201,7 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
   impl->m_kvalue = tableDesc.TableKValue;
   impl->m_minLoadFactor = tableDesc.MinLoadFactor;
   impl->m_maxLoadFactor = tableDesc.MaxLoadFactor;
+  impl->m_single_user_mode = tableDesc.SingleUserMode;
 
   impl->m_indexType = (NdbDictionary::Index::Type)
     getApiConstant(tableDesc.TableType,
@@ -1525,6 +1536,8 @@ NdbDictInterface::createOrAlterTable(Ndb & ndb,
   Uint64 minRows =
     (((Uint64)tmpTab.MinRowsHigh) << 32) + tmpTab.MinRowsLow;
 
+  tmpTab.SingleUserMode = impl.m_single_user_mode;
+
   tmpTab.FragmentType = getKernelConstant(impl.m_fragmentType,
 					  fragmentTypeMapping,
 					  DictTabInfo::AllNodesSmallTable);

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -138,6 +138,7 @@ public:
   int m_maxLoadFactor;
   Uint16 m_keyLenInWords;
   Uint16 m_fragmentCount;
+  Uint8 m_single_user_mode;
 
   NdbDictionaryImpl * m_dictionary;
   NdbIndexImpl * m_index;

ndb/test/src/NDBT_Table.cpp

@@ -31,7 +31,7 @@ operator <<(class NdbOut& ndbout, const NDBT_Table & tab)
   ndbout << "Number of attributes: " <<  tab.getNoOfColumns() << endl;
   ndbout << "Number of primary keys: " <<  tab.getNoOfPrimaryKeys() << endl;
   ndbout << "Length of frm data: " << tab.getFrmLength() << endl;
-
+  ndbout << "SingleUserMode: " << tab.getSingleUserMode() << endl;
 
   //<< ((tab.getTupleKey() == TupleId) ? " tupleid" : "") <<endl;
   ndbout << "TableStatus: ";