feature/qnode

include/common/tmsg.h

@@ -1840,3 +1840,4 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p
 #endif
 
 #endif /*_TD_COMMON_TAOS_MSG_H_*/
+

include/libs/catalog/catalog.h

@@ -30,7 +30,7 @@ extern "C" {
 #include "tmsg.h"
 #include "transport.h"
 
-struct SCatalog;
+typedef struct SCatalog SCatalog;
 
 enum {
   CTG_DBG_DB_NUM = 1,
@@ -64,6 +64,7 @@ typedef struct SCatalogCfg {
 typedef struct SSTableMetaVersion {
   char     dbFName[TSDB_DB_FNAME_LEN];
   char     stbName[TSDB_TABLE_NAME_LEN];
+  uint64_t dbId;
   uint64_t suid;
   int16_t  sversion;
   int16_t  tversion;  
@@ -84,7 +85,7 @@ int32_t catalogInit(SCatalogCfg *cfg);
  * @param catalogHandle (output, NO need to free it)
  * @return error code
  */
-int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
+int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle);
 
 /**
  * Free a cluster's all catalog info, usually it's not necessary, until the application is closing. 
@@ -92,9 +93,9 @@ int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
  * @param pCatalog (input, NO more usage)
  * @return error code
  */
-void catalogFreeHandle(struct SCatalog* pCatalog);
+void catalogFreeHandle(SCatalog* pCatalog);
 
-int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
+int32_t catalogGetDBVgVersion(SCatalog* pCatalog, const char* dbName, int32_t* version);
 
 /**
  * Get a DB's all vgroup info.
@@ -106,13 +107,13 @@ int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName,
  * @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
  * @return error code
  */
-int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
+int32_t catalogGetDBVgInfo(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
 
-int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, uint64_t dbId, SDBVgroupInfo* dbInfo);
+int32_t catalogUpdateDBVgInfo(SCatalog* pCatalog, const char* dbName, uint64_t dbId, SDBVgInfo* dbInfo);
 
-int32_t catalogRemoveDB(struct SCatalog* pCatalog, const char* dbName, uint64_t dbId);
+int32_t catalogRemoveDB(SCatalog* pCatalog, const char* dbName, uint64_t dbId);
 
-int32_t catalogRemoveSTableMeta(struct SCatalog* pCatalog, const char* dbName, const char* stbName, uint64_t suid);
+int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid);
 
 /**
  * Get a table's meta data. 
@@ -123,7 +124,7 @@ int32_t catalogRemoveSTableMeta(struct SCatalog* pCatalog, const char* dbName, c
  * @param pTableMeta(output, table meta data, NEED to free it by calller)
  * @return error code
  */
-int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
+int32_t catalogGetTableMeta(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
 
 /**
  * Get a super table's meta data. 
@@ -134,13 +135,13 @@ int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, cons
  * @param pTableMeta(output, table meta data, NEED to free it by calller)
  * @return error code
  */
-int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
+int32_t catalogGetSTableMeta(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
 
-int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg);
+int32_t catalogUpdateSTableMeta(SCatalog* pCatalog, STableMetaRsp *rspMsg);
 
 
 /**
- * Force renew a table's local cached meta data. 
+ * Force refresh a table's local cached meta data. 
  * @param pCatalog (input, got with catalogGetHandle)
  * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
@@ -148,10 +149,10 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
  * @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure) 
  * @return error code
  */
-  int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable);
+  int32_t catalogRefreshTableMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable);
 
 /**
- * Force renew a table's local cached meta data and get the new one. 
+ * Force refresh a table's local cached meta data and get the new one. 
  * @param pCatalog (input, got with catalogGetHandle)
  * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
@@ -160,7 +161,7 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
  * @param isSTable (input, is super table or not, 1:supposed to be stable, 0: supposed not to be stable, -1:not sure) 
  * @return error code
  */
-  int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
+  int32_t catalogRefreshGetTableMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
 
 
 
@@ -173,7 +174,7 @@ int32_t catalogUpdateSTableMeta(struct SCatalog* pCatalog, STableMetaRsp *rspMsg
  * @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
  * @return error code
  */
-int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
+int32_t catalogGetTableDistVgInfo(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
 
 /**
  * Get a table's vgroup from its name's hash value.
@@ -184,7 +185,7 @@ int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter,
  * @param vgInfo (output, vgroup info)
  * @return error code
  */
-int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pName, SVgroupInfo* vgInfo);
+int32_t catalogGetTableHashVgroup(SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pName, SVgroupInfo* vgInfo);
 
 
 /**
@@ -196,14 +197,14 @@ int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter
  * @param pRsp (output, response data)
  * @return error code 
  */
-int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
+int32_t catalogGetAllMeta(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
 
 
-int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
+int32_t catalogGetQnodeList(SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
 
-int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
+int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
 
-int32_t catalogGetExpiredDBs(struct SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
+int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
 
 
 /**

include/libs/qcom/query.h

@@ -80,16 +80,16 @@ typedef struct STableMeta {
   SSchema        schema[];
 } STableMeta;
 
-typedef struct SDBVgroupInfo {
+typedef struct SDBVgInfo {
   int32_t   vgVersion;  
   int8_t    hashMethod;
   SHashObj *vgHash;  //key:vgId, value:SVgroupInfo
-} SDBVgroupInfo;
+} SDBVgInfo;
 
 typedef struct SUseDbOutput {
   char           db[TSDB_DB_FNAME_LEN];
   uint64_t       dbId;
-  SDBVgroupInfo *dbVgroup;
+  SDBVgInfo     *dbVgroup;
 } SUseDbOutput;
 
 enum {

source/client/src/clientHb.c

@@ -43,7 +43,7 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
     if (rsp->vgVersion < 0) {
       code = catalogRemoveDB(pCatalog, rsp->db, rsp->uid);
     } else {
-      SDBVgroupInfo vgInfo = {0};
+      SDBVgInfo vgInfo = {0};
       vgInfo.vgVersion = rsp->vgVersion;
       vgInfo.hashMethod = rsp->hashMethod;
       vgInfo.vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
@@ -68,10 +68,7 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
         }
       }  
       
-      code = catalogUpdateDBVgroup(pCatalog, rsp->db, rsp->uid, &vgInfo);
+      catalogUpdateDBVgInfo(pCatalog, rsp->db, rsp->uid, &vgInfo);
-      if (code) {
-        taosHashCleanup(vgInfo.vgHash);
-      }
     }
 
     if (code) {
@@ -94,13 +91,14 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
 
     rsp->numOfColumns = ntohl(rsp->numOfColumns);
     rsp->suid = be64toh(rsp->suid);
+    rsp->dbId = be64toh(rsp->dbId);
     
     if (rsp->numOfColumns < 0) {
       schemaNum = 0;
       
       tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
 
-      catalogRemoveSTableMeta(pCatalog, rsp->dbFName, rsp->stbName, rsp->suid);
+      catalogRemoveStbMeta(pCatalog, rsp->dbFName, rsp->dbId, rsp->stbName, rsp->suid);
     } else {
       tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
 

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -382,7 +382,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
              pMgmt->openVnodes, pMgmt->totalVnodes);
     dndReportStartup(pDnode, "open-vnodes", stepDesc);
 
-    SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId};
+    SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId, .dbId = pCfg->dbUid};
     SVnode *  pImpl = vnodeOpen(pCfg->path, &cfg);
     if (pImpl == NULL) {
       dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
@@ -594,6 +594,7 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
 
   vnodeCfg.pDnode = pDnode;
   vnodeCfg.pTfs = pDnode->pTfs;
+  vnodeCfg.dbId = wrapperCfg.dbUid;
   SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
   if (pImpl == NULL) {
     dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());

source/dnode/mnode/impl/src/mndStb.c

@@ -1469,3 +1469,4 @@ static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
 }
+

source/dnode/vnode/inc/vnode.h

@@ -42,6 +42,7 @@ typedef struct STqCfg {
 
 typedef struct SVnodeCfg {
   int32_t  vgId;
+  uint64_t dbId;
   SDnode  *pDnode;
   STfs    *pTfs;
   uint64_t wsize;

source/dnode/vnode/src/vnd/vnodeMain.c

@@ -29,6 +29,7 @@ SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
     cfg.vgId = pVnodeCfg->vgId;
     cfg.pDnode = pVnodeCfg->pDnode;
     cfg.pTfs = pVnodeCfg->pTfs;
+    cfg.dbId = pVnodeCfg->dbId;
   }
 
   // Validate options

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -93,6 +93,7 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   if (pTbCfg->type == META_CHILD_TABLE) {
     pStbCfg = metaGetTbInfoByUid(pVnode->pMeta, pTbCfg->ctbCfg.suid);
     if (pStbCfg == NULL) {
+      code = TSDB_CODE_VND_TB_NOT_EXIST;
       goto _exit;
     }
 
@@ -116,9 +117,11 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   msgLen = sizeof(STableMetaRsp) + sizeof(SSchema) * (nCols + nTagCols);
   pTbMetaMsg = (STableMetaRsp *)rpcMallocCont(msgLen);
   if (pTbMetaMsg == NULL) {
+    code = TSDB_CODE_VND_OUT_OF_MEMORY;
     goto _exit;
   }
 
+  pTbMetaMsg->dbId = htobe64(pVnode->config.dbId);
   memcpy(pTbMetaMsg->dbFName, pReq->dbFName, sizeof(pTbMetaMsg->dbFName));
   strcpy(pTbMetaMsg->tbName, pReq->tbName);
   if (pTbCfg->type == META_CHILD_TABLE) {

source/libs/catalog/inc/catalogInt.h

@@ -27,7 +27,7 @@ extern "C" {
 #define CTG_DEFAULT_CACHE_CLUSTER_NUMBER 6
 #define CTG_DEFAULT_CACHE_VGROUP_NUMBER 100
 #define CTG_DEFAULT_CACHE_DB_NUMBER 20
-#define CTG_DEFAULT_CACHE_TABLEMETA_NUMBER 10000
+#define CTG_DEFAULT_CACHE_TBLMETA_NUMBER 1000
 #define CTG_DEFAULT_RENT_SECOND 10
 #define CTG_DEFAULT_RENT_SLOT_SIZE 10
 
@@ -47,9 +47,19 @@ enum {
   CTG_RENT_STABLE,
 };
 
+enum {
+  CTG_ACT_UPDATE_VG = 0,
+  CTG_ACT_UPDATE_TBL,
+  CTG_ACT_REMOVE_DB,
+  CTG_ACT_REMOVE_STB,
+  CTG_ACT_REMOVE_TBL,
+  CTG_ACT_MAX
+};
+
 typedef struct SCtgDebug {
   bool     lockDebug;
   bool     cacheDebug;
+  bool     apiDebug;
   uint32_t showCachePeriodSec;
 } SCtgDebug;
 
@@ -65,7 +75,7 @@ typedef struct SCtgDBCache {
   SRWLatch         vgLock;
   uint64_t         dbId;
   int8_t           deleted;
-  SDBVgroupInfo   *vgInfo;  
+  SDBVgInfo       *vgInfo;  
   SCtgTbMetaCache  tbCache;
 } SCtgDBCache;
 
@@ -85,7 +95,6 @@ typedef struct SCtgRentMgmt {
 
 typedef struct SCatalog {
   uint64_t         clusterId;  
-  SRWLatch         dbLock;
   SHashObj        *dbCache;      //key:dbname, value:SCtgDBCache
   SCtgRentMgmt     dbRent;
   SCtgRentMgmt     stbRent;
@@ -109,15 +118,57 @@ typedef struct SCatalogStat {
   SCtgCacheStat    cache;
 } SCatalogStat;
 
+typedef struct SCtgUpdateVgMsg {
+  SCatalog* pCtg;
+  char  dbFName[TSDB_DB_FNAME_LEN];
+  uint64_t dbId;
+  SDBVgInfo* dbInfo;
+} SCtgUpdateVgMsg;
+
+typedef struct SCtgUpdateTblMsg {
+  SCatalog* pCtg;
+  STableMetaOutput* output;
+} SCtgUpdateTblMsg;
+
+typedef struct SCtgRemoveDBMsg {
+  SCatalog* pCtg;
+  char  dbFName[TSDB_DB_FNAME_LEN];
+  uint64_t dbId;
+} SCtgRemoveDBMsg;
+
+typedef struct SCtgRemoveStbMsg {
+  SCatalog* pCtg;
+  char  dbFName[TSDB_DB_FNAME_LEN];
+  char  stbName[TSDB_TABLE_NAME_LEN];
+  uint64_t dbId;
+  uint64_t suid;
+} SCtgRemoveStbMsg;
+
+typedef struct SCtgMetaAction {
+  int32_t act;
+  void   *data;
+} SCtgMetaAction;
+
+typedef struct SCtgQNode {
+  SCtgMetaAction         action;
+  struct SCtgQNode      *next;
+} SCtgQNode;
+
 typedef struct SCatalogMgmt {
   bool                  exit;
   SRWLatch              lock;
+  SRWLatch              qlock;
+  SCtgQNode            *head;
+  SCtgQNode            *tail;
+  tsem_t                sem;  
+  pthread_t             updateThread;  
   SHashObj             *pCluster;     //key: clusterId, value: SCatalog*
   SCatalogStat          stat;
   SCatalogCfg           cfg;
 } SCatalogMgmt;
 
 typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
+typedef int32_t (*ctgActFunc)(SCtgMetaAction *);
 
 #define CTG_IS_META_NULL(type) ((type) == META_TYPE_NULL_TABLE)
 #define CTG_IS_META_CTABLE(type) ((type) == META_TYPE_CTABLE)
@@ -130,17 +181,21 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 #define CTG_SET_STABLE(isSTable, tbType) do { (isSTable) = ((tbType) == TSDB_SUPER_TABLE) ? 1 : ((tbType) > TSDB_SUPER_TABLE ? 0 : -1); } while (0)
 #define CTG_TBTYPE_MATCH(isSTable, tbType) (CTG_IS_UNKNOWN_STABLE(isSTable) || (CTG_IS_STABLE(isSTable) && (tbType) == TSDB_SUPER_TABLE) || (CTG_IS_NOT_STABLE(isSTable) && (tbType) != TSDB_SUPER_TABLE))
 
-#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
+#define CTG_META_SIZE(pMeta) (sizeof(STableMeta) + ((pMeta)->tableInfo.numOfTags + (pMeta)->tableInfo.numOfColumns) * sizeof(SSchema))
 
-#define ctgFatal(param, ...)  qFatal("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
-#define ctgError(param, ...)  qError("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define CTG_DB_NOT_EXIST(code) (code == TSDB_CODE_MND_DB_NOT_EXIST) 
-#define ctgWarn(param, ...)   qWarn("CTG:%p " param, pCatalog, __VA_ARGS__)
+
-#define ctgInfo(param, ...)   qInfo("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgFatal(param, ...)  qFatal("CTG:%p " param, pCtg, __VA_ARGS__)
-#define ctgDebug(param, ...)  qDebug("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgError(param, ...)  qError("CTG:%p " param, pCtg, __VA_ARGS__)
-#define ctgTrace(param, ...)  qTrace("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgWarn(param, ...)   qWarn("CTG:%p " param, pCtg, __VA_ARGS__)
+#define ctgInfo(param, ...)   qInfo("CTG:%p " param, pCtg, __VA_ARGS__)
+#define ctgDebug(param, ...)  qDebug("CTG:%p " param, pCtg, __VA_ARGS__)
+#define ctgTrace(param, ...)  qTrace("CTG:%p " param, pCtg, __VA_ARGS__)
 
 #define CTG_LOCK_DEBUG(...) do { if (gCTGDebug.lockDebug) { qDebug(__VA_ARGS__); } } while (0)
 #define CTG_CACHE_DEBUG(...) do { if (gCTGDebug.cacheDebug) { qDebug(__VA_ARGS__); } } while (0)
+#define CTG_API_DEBUG(...) do { if (gCTGDebug.apiDebug) { qDebug(__VA_ARGS__); } } while (0)
 
 #define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
 
@@ -181,8 +236,8 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 #define CTG_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
 #define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
 
-#define CTG_API_ENTER() do { CTG_LOCK(CTG_READ, &ctgMgmt.lock); if (atomic_load_8(&ctgMgmt.exit)) { CTG_RET(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
+#define CTG_API_ENTER() do { CTG_API_DEBUG("enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &ctgMgmt.lock); if (atomic_load_8(&ctgMgmt.exit)) { CTG_UNLOCK(CTG_READ, &ctgMgmt.lock); CTG_RET(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
-#define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &ctgMgmt.lock); CTG_RET(__code); } while (0)
+#define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &ctgMgmt.lock); CTG_API_DEBUG("leave %s", __FUNCTION__); CTG_RET(__code); } while (0)
 
 
 

source/libs/catalog/test/catalogTests.cpp

@@ -38,7 +38,6 @@ namespace {
 
 extern "C" int32_t ctgGetTableMetaFromCache(struct SCatalog *pCatalog, const SName *pTableName, STableMeta **pTableMeta,
                                             int32_t *exist);
-extern "C" int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output);
 extern "C" int32_t ctgDbgGetClusterCacheNum(struct SCatalog* pCatalog, int32_t type);
 
 void ctgTestSetPrepareTableMeta();
@@ -176,11 +175,11 @@ void ctgTestBuildCTableMetaOutput(STableMetaOutput *output) {
   strcpy(s->name, "tag1s");
 }
 
-void ctgTestBuildDBVgroup(SDBVgroupInfo **pdbVgroup) {
+void ctgTestBuildDBVgroup(SDBVgInfo **pdbVgroup) {
   static int32_t vgVersion = ctgTestVgVersion + 1;
   int32_t        vgNum = 0;
   SVgroupInfo    vgInfo = {0};
-  SDBVgroupInfo *dbVgroup = (SDBVgroupInfo *)calloc(1, sizeof(SDBVgroupInfo));
+  SDBVgInfo *dbVgroup = (SDBVgInfo *)calloc(1, sizeof(SDBVgInfo));
 
   dbVgroup->vgVersion = vgVersion++;
 
@@ -612,7 +611,7 @@ void *ctgTestGetDbVgroupThread(void *param) {
   int32_t          n = 0;
 
   while (!ctgTestStop) {
-    code = catalogGetDBVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
+    code = catalogGetDBVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
     if (code) {
       assert(0);
     }
@@ -635,12 +634,12 @@ void *ctgTestGetDbVgroupThread(void *param) {
 void *ctgTestSetSameDbVgroupThread(void *param) {
   struct SCatalog *pCtg = (struct SCatalog *)param;
   int32_t          code = 0;
-  SDBVgroupInfo    *dbVgroup = NULL;
+  SDBVgInfo       *dbVgroup = NULL;
   int32_t          n = 0;
 
   while (!ctgTestStop) {
     ctgTestBuildDBVgroup(&dbVgroup);
-    code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
+    code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
     if (code) {
       assert(0);
     }
@@ -660,12 +659,12 @@ void *ctgTestSetSameDbVgroupThread(void *param) {
 void *ctgTestSetDiffDbVgroupThread(void *param) {
   struct SCatalog *pCtg = (struct SCatalog *)param;
   int32_t          code = 0;
-  SDBVgroupInfo    *dbVgroup = NULL;
+  SDBVgInfo    *dbVgroup = NULL;
   int32_t          n = 0;
 
   while (!ctgTestStop) {
     ctgTestBuildDBVgroup(&dbVgroup);
-    code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId++, dbVgroup);
+    code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId++, dbVgroup);
     if (code) {
       assert(0);
     }
@@ -716,14 +715,22 @@ void *ctgTestGetCtableMetaThread(void *param) {
 void *ctgTestSetCtableMetaThread(void *param) {
   struct SCatalog *pCtg = (struct SCatalog *)param;
   int32_t          code = 0;
-  SDBVgroupInfo    dbVgroup = {0};
+  SDBVgInfo    dbVgroup = {0};
   int32_t          n = 0;
   STableMetaOutput output = {0};
 
   ctgTestBuildCTableMetaOutput(&output);
+  SCtgMetaAction action = {0};
+  
+  action.act = CTG_ACT_UPDATE_TBL;
 
   while (!ctgTestStop) {
-    code = ctgUpdateTableMetaCache(pCtg, &output);
+    SCtgUpdateTblMsg *msg = malloc(sizeof(SCtgUpdateTblMsg));
+    msg->pCtg = pCtg;
+    msg->output = output;
+    action.data = msg;
+
+    code = ctgActUpdateTbl(&action);
     if (code) {
       assert(0);
     }
@@ -984,7 +991,7 @@ TEST(tableMeta, superTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
   tableMeta = NULL;
-  code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
+  code = catalogRefreshGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(tableMeta->vgId, 9);
   ASSERT_EQ(tableMeta->tableType, TSDB_CHILD_TABLE);
@@ -1174,7 +1181,7 @@ TEST(tableDistVgroup, normalTable) {
   strcpy(n.dbname, "db1");
   strcpy(n.tname, ctgTestTablename);
 
-  code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
+  code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
   vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@@ -1206,7 +1213,7 @@ TEST(tableDistVgroup, childTableCase) {
   strcpy(n.dbname, "db1");
   strcpy(n.tname, ctgTestCTablename);
 
-  code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
+  code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
   vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@@ -1237,7 +1244,7 @@ TEST(tableDistVgroup, superTableCase) {
   strcpy(n.dbname, "db1");
   strcpy(n.tname, ctgTestSTablename);
 
-  code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
+  code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 10);
   vgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@@ -1258,7 +1265,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   void            *mockPointer = (void *)0x1;
   SVgroupInfo      vgInfo = {0};
   SVgroupInfo     *pvgInfo = NULL;
-  SDBVgroupInfo    *dbVgroup = NULL;
+  SDBVgInfo    *dbVgroup = NULL;
   SArray          *vgList = NULL;
 
   ctgTestInitLogFile();
@@ -1279,7 +1286,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   strcpy(n.dbname, "db1");
   strcpy(n.tname, ctgTestTablename);
 
-  code = catalogGetDBVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
+  code = catalogGetDBVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, ctgTestDbname, false, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), ctgTestVgNum);
 
@@ -1288,7 +1295,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   ASSERT_EQ(vgInfo.vgId, 8);
   ASSERT_EQ(vgInfo.epset.numOfEps, 3);
 
-  code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
+  code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
   pvgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@@ -1297,7 +1304,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   taosArrayDestroy(vgList);
 
   ctgTestBuildDBVgroup(&dbVgroup);
-  code = catalogUpdateDBVgroup(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
+  code = catalogUpdateDBVgInfo(pCtg, ctgTestDbname, ctgTestDbId, dbVgroup);
   ASSERT_EQ(code, 0);
 
   code = catalogGetTableHashVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgInfo);
@@ -1305,7 +1312,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   ASSERT_EQ(vgInfo.vgId, 7);
   ASSERT_EQ(vgInfo.epset.numOfEps, 2);
 
-  code = catalogGetTableDistVgroup(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
+  code = catalogGetTableDistVgInfo(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &vgList);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), 1);
   pvgInfo = (SVgroupInfo *)taosArrayGet(vgList, 0);
@@ -1321,7 +1328,7 @@ TEST(multiThread, getSetRmSameDbVgroup) {
   void            *mockPointer = (void *)0x1;
   SVgroupInfo      vgInfo = {0};
   SVgroupInfo     *pvgInfo = NULL;
-  SDBVgroupInfo    dbVgroup = {0};
+  SDBVgInfo    dbVgroup = {0};
   SArray          *vgList = NULL;
   ctgTestStop = false;
 
@@ -1372,7 +1379,7 @@ TEST(multiThread, getSetRmDiffDbVgroup) {
   void            *mockPointer = (void *)0x1;
   SVgroupInfo      vgInfo = {0};
   SVgroupInfo     *pvgInfo = NULL;
-  SDBVgroupInfo    dbVgroup = {0};
+  SDBVgInfo    dbVgroup = {0};
   SArray          *vgList = NULL;
   ctgTestStop = false;
 
@@ -1425,7 +1432,7 @@ TEST(multiThread, ctableMeta) {
   void            *mockPointer = (void *)0x1;
   SVgroupInfo      vgInfo = {0};
   SVgroupInfo     *pvgInfo = NULL;
-  SDBVgroupInfo    dbVgroup = {0};
+  SDBVgInfo    dbVgroup = {0};
   SArray          *vgList = NULL;
   ctgTestStop = false;
 
@@ -1477,7 +1484,7 @@ TEST(rentTest, allRent) {
   void            *mockPointer = (void *)0x1;
   SVgroupInfo      vgInfo = {0};
   SVgroupInfo     *pvgInfo = NULL;
-  SDBVgroupInfo    dbVgroup = {0};
+  SDBVgInfo    dbVgroup = {0};
   SArray          *vgList = NULL;
   ctgTestStop = false;
   SDbVgVersion       *dbs = NULL;

source/libs/parser/src/astValidate.c

@@ -3647,7 +3647,7 @@ int32_t evaluateSqlNode(SSqlNode* pNode, int32_t tsPrecision, SMsgBuf* pMsgBuf)
 //TODO remove it
 int32_t setTableVgroupList(SParseContext *pCtx, SName* name, SVgroupsInfo **pVgList) {
   SArray* vgroupList = NULL;
-  int32_t code = catalogGetTableDistVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, name, &vgroupList);
+  int32_t code = catalogGetTableDistVgInfo(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, name, &vgroupList);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }

source/libs/parser/src/dCDAstProcess.c

@@ -50,7 +50,7 @@ static int32_t setShowInfo(SShowInfo* pShowInfo, SParseContext* pCtx, void** out
     char dbFname[TSDB_DB_FNAME_LEN] = {0};
     tNameGetFullDbName(&name, dbFname);
 
-    int32_t code = catalogGetDBVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, dbFname, false, &array);
+    int32_t code = catalogGetDBVgInfo(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, dbFname, false, &array);
     if (code != TSDB_CODE_SUCCESS) {
       terrno = code;
       return code;

source/libs/qcom/src/querymsg.c

@@ -113,9 +113,9 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
     return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
   }
 
-  pOut->dbVgroup = calloc(1, sizeof(SDBVgroupInfo));
+  pOut->dbVgroup = calloc(1, sizeof(SDBVgInfo));
   if (NULL == pOut->dbVgroup) {
-    qError("calloc %d failed", (int32_t)sizeof(SDBVgroupInfo));
+    qError("calloc %d failed", (int32_t)sizeof(SDBVgInfo));
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }