Merge pull request #2713 from taosdata/feature/vnode

Feature/vnode

src/mnode/src/mnodeSdb.c

@@ -77,11 +77,6 @@ typedef struct {
   pthread_mutex_t mutex;
 } SSdbObject;
 
-typedef struct {
-  int32_t rowSize;
-  void *  row;
-} SSdbRow;
-
 typedef struct {
   pthread_t thread;
   int32_t   workerId;
@@ -419,32 +414,28 @@ void sdbDecRef(void *handle, void *pObj) {
   }
 }
 
-static SSdbRow *sdbGetRowMeta(SSdbTable *pTable, void *key) {
+static void *sdbGetRowMeta(SSdbTable *pTable, void *key) {
   if (pTable == NULL) return NULL;
 
   int32_t keySize = sizeof(int32_t);
   if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
     keySize = strlen((char *)key);
   }
-  
-  return taosHashGet(pTable->iHandle, key, keySize);
+
+  void **ppRow = (void **)taosHashGet(pTable->iHandle, key, keySize);
+  if (ppRow == NULL) return NULL;
+  return *ppRow;
 }
 
-static SSdbRow *sdbGetRowMetaFromObj(SSdbTable *pTable, void *key) {
+static void *sdbGetRowMetaFromObj(SSdbTable *pTable, void *key) {
   return sdbGetRowMeta(pTable, sdbGetObjKey(pTable, key));
 }
 
 void *sdbGetRow(void *handle, void *key) {
-  SSdbTable *pTable = (SSdbTable *)handle;
-  int32_t keySize = sizeof(int32_t);
-  if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
-    keySize = strlen((char *)key);
-  }
-  
-  SSdbRow *pMeta = taosHashGet(pTable->iHandle, key, keySize);
-  if (pMeta) {
-    sdbIncRef(pTable, pMeta->row);
-    return pMeta->row;
+  void *pRow = sdbGetRowMeta(handle, key);
+  if (pRow) {
+    sdbIncRef(handle, pRow);
+    return pRow;
   } else {
     return NULL;
   }
@@ -455,10 +446,6 @@ static void *sdbGetRowFromObj(SSdbTable *pTable, void *key) {
 }
 
 static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
-  SSdbRow rowMeta;
-  rowMeta.rowSize = pOper->rowSize;
-  rowMeta.row = pOper->pObj;
-
   void *  key = sdbGetObjKey(pTable, pOper->pObj);
   int32_t keySize = sizeof(int32_t);
 
@@ -466,7 +453,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
     keySize = strlen((char *)key);
   }
 
-  taosHashPut(pTable->iHandle, key, keySize, &rowMeta, sizeof(SSdbRow));
+  taosHashPut(pTable->iHandle, key, keySize, &pOper->pObj, sizeof(void **));
 
   sdbIncRef(pTable, pOper->pObj);
   atomic_add_fetch_32(&pTable->numOfRows, 1);
@@ -586,17 +573,17 @@ static int sdbWrite(void *param, void *data, int type) {
     code = (*pTable->decodeFp)(&oper);
     return sdbInsertHash(pTable, &oper);
   } else if (action == SDB_ACTION_DELETE) {
-    SSdbRow *rowMeta = sdbGetRowMeta(pTable, pHead->cont);
-    if (rowMeta == NULL || rowMeta->row == NULL) {
+    void *pRow = sdbGetRowMeta(pTable, pHead->cont);
+    if (pRow == NULL) {
       sdbError("table:%s, failed to get object:%s from wal while dispose delete action", pTable->tableName,
                pHead->cont);
       return TSDB_CODE_SUCCESS;
     }
-    SSdbOper oper = {.table = pTable, .pObj = rowMeta->row};
+    SSdbOper oper = {.table = pTable, .pObj = pRow};
     return sdbDeleteHash(pTable, &oper);
   } else if (action == SDB_ACTION_UPDATE) {
-    SSdbRow *rowMeta = sdbGetRowMeta(pTable, pHead->cont);
-    if (rowMeta == NULL || rowMeta->row == NULL) {
+    void *pRow = sdbGetRowMeta(pTable, pHead->cont);
+    if (pRow == NULL) {
       sdbError("table:%s, failed to get object:%s from wal while dispose update action", pTable->tableName,
                pHead->cont);
       return TSDB_CODE_SUCCESS;
@@ -675,18 +662,12 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
   if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
-  SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
-  if (pMeta == NULL) {
+  void *pRow = sdbGetRowMetaFromObj(pTable, pOper->pObj);
+  if (pRow == NULL) {
     sdbDebug("table:%s, record is not there, delete failed", pTable->tableName);
     return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
   }
 
-  void *pMetaRow = pMeta->row;
-  if (pMetaRow == NULL) {
-    sdbError("table:%s, record meta is null", pTable->tableName);
-    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
-  }
-
   sdbIncRef(pTable, pOper->pObj);
 
   int32_t code = sdbDeleteHash(pTable, pOper);
@@ -728,18 +709,12 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
   if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
-  SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
-  if (pMeta == NULL) {
+  void *pRow = sdbGetRowMetaFromObj(pTable, pOper->pObj);
+  if (pRow == NULL) {
     sdbDebug("table:%s, record is not there, update failed", pTable->tableName);
     return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
   }
 
-  void *pMetaRow = pMeta->row;
-  if (pMetaRow == NULL) {
-    sdbError("table:%s, record meta is null", pTable->tableName);
-    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
-  }
-
   int32_t code = sdbUpdateHash(pTable, pOper);
   if (code != TSDB_CODE_SUCCESS) {
     sdbError("table:%s, failed to update hash", pTable->tableName);
@@ -789,14 +764,14 @@ void *sdbFetchRow(void *handle, void *pNode, void **ppRow) {
     return NULL;
   }
 
-  SSdbRow *pMeta = taosHashIterGet(pIter);
-  if (pMeta == NULL) {
+  void **ppMetaRow = taosHashIterGet(pIter);
+  if (ppMetaRow == NULL) {
     taosHashDestroyIter(pIter);
     return NULL;
   }
 
-  *ppRow = pMeta->row;
-  sdbIncRef(handle, pMeta->row);
+  *ppRow = *ppMetaRow;
+  sdbIncRef(handle, *ppMetaRow);
 
   return pIter;
 }
@@ -846,11 +821,11 @@ void sdbCloseTable(void *handle) {
 
   SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
   while (taosHashIterNext(pIter)) {
-    SSdbRow *pMeta = taosHashIterGet(pIter);
-    if (pMeta == NULL) continue;
+    void **ppRow = taosHashIterGet(pIter);
+    if (ppRow == NULL) continue;
 
     SSdbOper oper = {
-      .pObj = pMeta->row,
+      .pObj = *ppRow,
       .table = pTable,
     };
 

src/mnode/src/mnodeTable.c

@@ -107,42 +107,41 @@ static int32_t mnodeChildTableActionInsert(SSdbOper *pOper) {
   SVgObj *pVgroup = mnodeGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
     mError("ctable:%s, not in vgId:%d", pTable->info.tableId, pTable->vgId);
-    return TSDB_CODE_MND_VGROUP_NOT_EXIST;
   }
-  mnodeDecVgroupRef(pVgroup);
 
-  SDbObj *pDb = mnodeGetDb(pVgroup->dbName);
-  if (pDb == NULL) {
-    mError("ctable:%s, vgId:%d not in db:%s", pTable->info.tableId, pVgroup->vgId, pVgroup->dbName);
-    return TSDB_CODE_MND_INVALID_DB;
-  }
-  
-  if (pDb->status != TSDB_DB_STATUS_READY) {
-    mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
-    return TSDB_CODE_MND_DB_IN_DROPPING;
+  SDbObj *pDb = NULL;
+  if (pVgroup != NULL) {
+    pDb = mnodeGetDb(pVgroup->dbName);
+    if (pDb == NULL) {
+      mError("ctable:%s, vgId:%d not in db:%s", pTable->info.tableId, pVgroup->vgId, pVgroup->dbName);
+    }
   }
-  mnodeDecDbRef(pDb);
 
-  SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
-  if (pAcct == NULL) {
-    mError("ctable:%s, acct:%s not exists", pTable->info.tableId, pDb->acct);
-    return TSDB_CODE_MND_INVALID_ACCT;
+  SAcctObj *pAcct = NULL;
+  if (pDb != NULL) {
+    pAcct = mnodeGetAcct(pDb->acct);
+    if (pAcct == NULL) {
+      mError("ctable:%s, acct:%s not exists", pTable->info.tableId, pDb->acct);
+    }
   }
-  mnodeDecAcctRef(pAcct);
 
   if (pTable->info.type == TSDB_CHILD_TABLE) {
     // add ref
     pTable->superTable = mnodeGetSuperTableByUid(pTable->suid);
     mnodeAddTableIntoStable(pTable->superTable, pTable);
     grantAdd(TSDB_GRANT_TIMESERIES, pTable->superTable->numOfColumns - 1);
-    pAcct->acctInfo.numOfTimeSeries += (pTable->superTable->numOfColumns - 1);
+    if (pAcct) pAcct->acctInfo.numOfTimeSeries += (pTable->superTable->numOfColumns - 1);
   } else {
     grantAdd(TSDB_GRANT_TIMESERIES, pTable->numOfColumns - 1);
-    pAcct->acctInfo.numOfTimeSeries += (pTable->numOfColumns - 1);
+    if (pAcct) pAcct->acctInfo.numOfTimeSeries += (pTable->numOfColumns - 1);
   }
 
-  mnodeAddTableIntoDb(pDb);
-  mnodeAddTableIntoVgroup(pVgroup, pTable);
+  if (pDb) mnodeAddTableIntoDb(pDb);
+  if (pVgroup) mnodeAddTableIntoVgroup(pVgroup, pTable);
+
+  mnodeDecVgroupRef(pVgroup);
+  mnodeDecDbRef(pDb);
+  mnodeDecAcctRef(pAcct);
 
   return TSDB_CODE_SUCCESS;
 }