Merge pull request #9662 from taosdata/feature/qnode

Feature/qnode

include/common/tmsg.h

@@ -871,6 +871,7 @@ typedef struct {
 
 typedef struct {
   char        db[TSDB_DB_FNAME_LEN];
+  int64_t     uid;
   int32_t     vgVersion;
   int32_t     vgNum;
   int8_t      hashMethod;

include/libs/catalog/catalog.h

@@ -48,8 +48,22 @@ typedef struct SMetaData {
 typedef struct SCatalogCfg {
   uint32_t maxTblCacheNum;
   uint32_t maxDBCacheNum;
+  uint32_t dbRentSec;
+  uint32_t stableRentSec;
 } SCatalogCfg;
 
+typedef struct SSTableMetaVersion {
+  uint64_t suid;
+  int16_t  sversion;
+  int16_t  tversion;  
+} SSTableMetaVersion;
+
+typedef struct SDbVgVersion {
+  int64_t dbId;
+  int32_t vgVersion;
+} SDbVgVersion;
+
+
 int32_t catalogInit(SCatalogCfg *cfg);
 
 /**
@@ -60,6 +74,14 @@ int32_t catalogInit(SCatalogCfg *cfg);
  */
 int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
 
+/**
+ * Free a cluster's all catalog info, usually it's not necessary, until the application is closing. 
+ * no current or future usage should be guaranteed by application
+ * @param pCatalog (input, NO more usage)
+ * @return error code
+ */
+void catalogFreeHandle(struct SCatalog* pCatalog);
+
 int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
 
 /**
@@ -87,15 +109,28 @@ int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDB
  */
 int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta);
 
+/**
+ * Get a super table's meta data. 
+ * @param pCatalog (input, got with catalogGetHandle)
+ * @param pTransporter (input, rpc object)
+ * @param pMgmtEps (input, mnode EPs)
+ * @param pTableName (input, table name, NOT including db name)
+ * @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);
+
+
 /**
  * Force renew a table's local cached meta data. 
  * @param pCatalog (input, got with catalogGetHandle)
  * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
  * @param pTableName (input, table name, NOT including db name)
+ * @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 catalogRenewTableMeta(struct 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. 
@@ -104,9 +139,11 @@ int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void * pTransporter, co
  * @param pMgmtEps (input, mnode EPs)
  * @param pTableName (input, table name, NOT including db name)
  * @param pTableMeta(output, table meta data, NEED to free it by calller) 
+ * @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 catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable);
+
 
 
 /**
@@ -146,6 +183,9 @@ int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* p
 
 int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList);
 
+int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
+
+int32_t catalogGetExpiredDBs(struct SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num);
 
 
 /**

include/libs/qcom/query.h

@@ -76,6 +76,7 @@ typedef struct STableMeta {
 
 typedef struct SDBVgroupInfo {
   SRWLatch  lock;
+  int64_t   dbId;
   int32_t   vgVersion;  
   int8_t    hashMethod;
   SHashObj *vgInfo;  //key:vgId, value:SVgroupInfo
@@ -86,8 +87,15 @@ typedef struct SUseDbOutput {
   SDBVgroupInfo dbVgroup;
 } SUseDbOutput;
 
+typedef enum {
+  META_TYPE_NON_TABLE = 1,
+  META_TYPE_CTABLE,
+  META_TYPE_TABLE,
+  META_TYPE_BOTH_TABLE,
+};
+
 typedef struct STableMetaOutput {
-  int32_t     metaNum;
+  int32_t     metaType;
   char        ctbFname[TSDB_TABLE_FNAME_LEN];
   char        tbFname[TSDB_TABLE_FNAME_LEN];
   SCTableMeta ctbMeta;
@@ -149,6 +157,11 @@ void initQueryModuleMsgHandle();
 extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen);
 extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char *msg, int32_t msgSize);
 
+#define SET_META_TYPE_NONE(t) (t) = META_TYPE_NON_TABLE
+#define SET_META_TYPE_CTABLE(t) (t) = META_TYPE_CTABLE
+#define SET_META_TYPE_TABLE(t) (t) = META_TYPE_TABLE
+#define SET_META_TYPE_BOTH_TABLE(t) (t) = META_TYPE_BOTH_TABLE
+
 #define qFatal(...)  do { if (qDebugFlag & DEBUG_FATAL) { taosPrintLog("QRY FATAL ", qDebugFlag, __VA_ARGS__); }} while(0)
 #define qError(...)  do { if (qDebugFlag & DEBUG_ERROR) { taosPrintLog("QRY ERROR ", qDebugFlag, __VA_ARGS__); }} while(0)
 #define qWarn(...)   do { if (qDebugFlag & DEBUG_WARN)  { taosPrintLog("QRY WARN ", qDebugFlag, __VA_ARGS__); }}  while(0)

include/util/thash.h

@@ -124,6 +124,9 @@ int32_t taosHashGetSize(const SHashObj *pHashObj);
  */
 int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size);
 
+int32_t taosHashPutExt(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size, bool *newAdded);
+
+
 /**
  * return the payload data with the specified key
  *

include/util/tlog.h

@@ -44,7 +44,6 @@ extern int32_t tsdbDebugFlag;
 extern int32_t tqDebugFlag;
 extern int32_t cqDebugFlag;
 extern int32_t debugFlag;
-extern int32_t ctgDebugFlag;
 
 #define DEBUG_FATAL 1U
 #define DEBUG_ERROR DEBUG_FATAL

source/client/src/clientMain.c

@@ -46,6 +46,7 @@ void taos_cleanup(void) {
   taosCloseRef(id);
 
   rpcCleanup();
+  catalogDestroy();
   taosCloseLog();
 
   tscInfo("all local resources released");

source/common/src/tglobal.c

@@ -927,16 +927,6 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosAddConfigOption(cfg);
 
-  cfg.option = "ctgDebugFlag";
-  cfg.ptr = &ctgDebugFlag;
-  cfg.valType = TAOS_CFG_VTYPE_INT32;
-  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
-  cfg.minValue = 0;
-  cfg.maxValue = 255;
-  cfg.ptrLength = 0;
-  cfg.unitType = TAOS_CFG_UTYPE_NONE;
-  taosAddConfigOption(cfg);
-
   cfg.option = "enableRecordSql";
   cfg.ptr = &tsTscEnableRecordSql;
   cfg.valType = TAOS_CFG_VTYPE_INT8;

source/dnode/mgmt/impl/test/stb/stb.cpp

@@ -137,9 +137,9 @@ TEST_F(DndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
     pRsp->numOfColumns = htonl(pRsp->numOfColumns);
     pRsp->sversion = htonl(pRsp->sversion);
     pRsp->tversion = htonl(pRsp->tversion);
-    pRsp->suid = htobe64(pRsp->suid);
-    pRsp->tuid = htobe64(pRsp->tuid);
-    pRsp->vgId = htobe64(pRsp->vgId);
+    pRsp->suid = be64toh(pRsp->suid);
+    pRsp->tuid = be64toh(pRsp->tuid);
+    pRsp->vgId = be64toh(pRsp->vgId);
     for (int32_t i = 0; i < pRsp->numOfTags + pRsp->numOfColumns; ++i) {
       SSchema* pSchema = &pRsp->pSchema[i];
       pSchema->colId = htonl(pSchema->colId);
@@ -156,7 +156,7 @@ TEST_F(DndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
     EXPECT_EQ(pRsp->sversion, 1);
     EXPECT_EQ(pRsp->tversion, 0);
     EXPECT_GT(pRsp->suid, 0);
-    EXPECT_EQ(pRsp->tuid, 0);
+    EXPECT_GT(pRsp->tuid, 0);
     EXPECT_EQ(pRsp->vgId, 0);
 
     {

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

@@ -917,6 +917,7 @@ static int32_t mndProcessUseDbMsg(SMnodeMsg *pMsg) {
   }
 
   memcpy(pRsp->db, pDb->name, TSDB_DB_FNAME_LEN);
+  pRsp->uid = htobe64(pDb->uid);
   pRsp->vgVersion = htonl(pDb->vgVersion);
   pRsp->vgNum = htonl(vindex);
   pRsp->hashMethod = pDb->hashMethod;

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

@@ -769,7 +769,8 @@ static int32_t mndProcessStbMetaMsg(SMnodeMsg *pMsg) {
   pMeta->tableType = TSDB_SUPER_TABLE;
   pMeta->update = pDb->cfg.update;
   pMeta->sversion = htonl(pStb->version);
-  pMeta->suid = htonl(pStb->uid);
+  pMeta->suid = htobe64(pStb->uid);
+  pMeta->tuid = htobe64(pStb->uid);
 
   for (int32_t i = 0; i < totalCols; ++i) {
     SSchema *pSchema = &pMeta->pSchema[i];

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

@@ -105,6 +105,9 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   if (pTbCfg->type == META_CHILD_TABLE) {
     strcpy(pTbMetaMsg->stbFname, pStbCfg->name);
     pTbMetaMsg->suid = htobe64(pTbCfg->ctbCfg.suid);
+  } else if (pTbCfg->type == META_SUPER_TABLE) {
+    strcpy(pTbMetaMsg->stbFname, pTbCfg->name);
+    pTbMetaMsg->suid = htobe64(uid);
   }
   pTbMetaMsg->numOfTags = htonl(nTagCols);
   pTbMetaMsg->numOfColumns = htonl(nCols);

source/libs/catalog/inc/catalogInt.h

@@ -22,20 +22,31 @@ extern "C" {
 
 #include "catalog.h"
 #include "common.h"
-#include "tlog.h"
+#include "query.h"
 
 #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 100000
+#define CTG_DEFAULT_RENT_SECOND 10
+#define CTG_DEFAULT_RENT_SLOT_SIZE 10
+
+#define CTG_RENT_SLOT_SECOND 2
 
 #define CTG_DEFAULT_INVALID_VERSION (-1)
 
+#define CTG_ERR_CODE_TABLE_NOT_EXIST TSDB_CODE_TDB_INVALID_TABLE_ID
+
 enum {
   CTG_READ = 1,
   CTG_WRITE,
 };
 
+enum {
+  CTG_RENT_DB = 1,
+  CTG_RENT_STABLE,
+};
+
 typedef struct SVgroupListCache {
   int32_t vgroupVersion;
   SHashObj *cache;        // key:vgId, value:SVgroupInfo
@@ -51,30 +62,76 @@ typedef struct STableMetaCache {
   SHashObj *stableCache;     //key:suid, value:STableMeta*
 } STableMetaCache;
 
+typedef struct SRentSlotInfo {
+  SRWLatch lock;
+  bool     needSort;
+  SArray  *meta;  // element is SDbVgVersion or SSTableMetaVersion
+} SRentSlotInfo;
+
+typedef struct SMetaRentMgmt {
+  int8_t         type;
+  uint16_t       slotNum;
+  uint16_t       slotRIdx;
+  int64_t        lastReadMsec;
+  SRentSlotInfo *slots;
+} SMetaRentMgmt;
+
 typedef struct SCatalog {
+  uint64_t         clusterId;
   SDBVgroupCache   dbCache;
   STableMetaCache  tableCache;
+  SMetaRentMgmt    dbRent;
+  SMetaRentMgmt    stableRent;
 } SCatalog;
 
+typedef struct SCtgApiStat {
+
+} SCtgApiStat;
+
+typedef struct SCtgResourceStat {
+
+} SCtgResourceStat;
+
+typedef struct SCtgCacheStat {
+
+} SCtgCacheStat;
+
+typedef struct SCatalogStat {
+  SCtgApiStat      api;
+  SCtgResourceStat resource;
+  SCtgCacheStat    cache;
+} SCatalogStat;
+
 typedef struct SCatalogMgmt {
-  void       *pMsgSender;   // used to send messsage to mnode to fetch necessary metadata
-  SHashObj   *pCluster;     // items cached for each cluster, the hash key is the cluster-id got from mgmt node
+  SHashObj             *pCluster;     //key: clusterId, value: SCatalog*
+  SCatalogStat          stat;
   SCatalogCfg           cfg;
 } SCatalogMgmt;
 
 typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 
-#define ctgFatal(...)  do { if (ctgDebugFlag & DEBUG_FATAL) { taosPrintLog("CTG FATAL ", ctgDebugFlag, __VA_ARGS__); }} while(0)
-#define ctgError(...)  do { if (ctgDebugFlag & DEBUG_ERROR) { taosPrintLog("CTG ERROR ", ctgDebugFlag, __VA_ARGS__); }} while(0)
-#define ctgWarn(...)   do { if (ctgDebugFlag & DEBUG_WARN)  { taosPrintLog("CTG WARN ", ctgDebugFlag, __VA_ARGS__); }}  while(0)
-#define ctgInfo(...)   do { if (ctgDebugFlag & DEBUG_INFO)  { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0)
-#define ctgDebug(...)  do { if (ctgDebugFlag & DEBUG_DEBUG) { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0)
-#define ctgTrace(...)  do { if (ctgDebugFlag & DEBUG_TRACE) { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0)
-#define ctgDebugL(...) do { if (ctgDebugFlag & DEBUG_DEBUG) { taosPrintLongString("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0)
+#define CTG_IS_META_NONE(type) ((type) == META_TYPE_NON_TABLE)
+#define CTG_IS_META_CTABLE(type) ((type) == META_TYPE_CTABLE)
+#define CTG_IS_META_TABLE(type) ((type) == META_TYPE_TABLE)
+#define CTG_IS_META_BOTH(type) ((type) == META_TYPE_BOTH_TABLE)
+
+#define CTG_IS_STABLE(isSTable) (1 == (isSTable))
+#define CTG_IS_NOT_STABLE(isSTable) (0 == (isSTable))
+#define CTG_IS_UNKNOWN_STABLE(isSTable) ((isSTable) < 0)
+#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 ctgFatal(param, ...)  qFatal("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgError(param, ...)  qError("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgWarn(param, ...)   qWarn("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgInfo(param, ...)   qInfo("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgDebug(param, ...)  qDebug("CTG:%p " param, pCatalog, __VA_ARGS__)
+#define ctgTrace(param, ...)  qTrace("CTG:%p " param, pCatalog, __VA_ARGS__)
 
 #define CTG_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
 #define CTG_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
-#define CTG_ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { ctgError(__VA_ARGS__); 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 TD_RWLATCH_WRITE_FLAG_COPY 0x40000000
@@ -82,15 +139,15 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 #define CTG_LOCK(type, _lock) do {   \
   if (CTG_READ == (type)) {          \
     assert(atomic_load_32((_lock)) >= 0);  \
-    ctgDebug("CTG RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     taosRLockLatch(_lock);           \
-    ctgDebug("CTG RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     assert(atomic_load_32((_lock)) > 0);  \
   } else {                                                \
     assert(atomic_load_32((_lock)) >= 0);  \
-    ctgDebug("CTG WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
+    qDebug("CTG WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
     taosWLockLatch(_lock);                                \
-    ctgDebug("CTG WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
+    qDebug("CTG WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
     assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
   }                                                       \
 } while (0)
@@ -98,15 +155,15 @@ typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 #define CTG_UNLOCK(type, _lock) do {                       \
   if (CTG_READ == (type)) {                                \
     assert(atomic_load_32((_lock)) > 0);  \
-    ctgDebug("CTG RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     taosRUnLockLatch(_lock);                              \
-    ctgDebug("CTG RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     assert(atomic_load_32((_lock)) >= 0);  \
   } else {                                                \
     assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
-    ctgDebug("CTG WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     taosWUnLockLatch(_lock);                              \
-    ctgDebug("CTG WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
+    qDebug("CTG WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
     assert(atomic_load_32((_lock)) >= 0);  \
   }                                                       \
 } while (0)

source/libs/catalog/test/CMakeLists.txt

@@ -16,3 +16,8 @@ TARGET_INCLUDE_DIRECTORIES(
         PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/catalog/"
         PRIVATE "${CMAKE_SOURCE_DIR}/source/libs/catalog/inc"
 )
+
+add_test(
+    NAME catalogTest
+    COMMAND catalogTest
+)

source/libs/catalog/test/catalogTests.cpp

@@ -42,10 +42,13 @@ extern "C" int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMeta
 void ctgTestSetPrepareTableMeta();
 void ctgTestSetPrepareCTableMeta();
 void ctgTestSetPrepareSTableMeta();
+void ctgTestSetPrepareMultiSTableMeta();
 
 bool ctgTestStop = false;
 bool ctgTestEnableSleep = false;
-bool ctgTestDeadLoop = true;
+bool ctgTestDeadLoop = false;
+int32_t ctgTestPrintNum = 200000;
+int32_t ctgTestMTRunSec = 30;
 
 int32_t ctgTestCurrentVgVersion = 0;
 int32_t ctgTestVgVersion = 1;
@@ -54,6 +57,8 @@ int32_t ctgTestColNum = 2;
 int32_t ctgTestTagNum = 1;
 int32_t ctgTestSVersion = 1;
 int32_t ctgTestTVersion = 1;
+int32_t ctgTestSuid = 2;
+int64_t ctgTestDbId = 33;
 
 uint64_t ctgTestClusterId = 0x1;
 char *ctgTestDbname = "1.db1";
@@ -101,7 +106,6 @@ void ctgTestInitLogFile() {
   const char    *defaultLogFileNamePrefix = "taoslog";
   const int32_t  maxLogFileNum = 10;
 
-  ctgDebugFlag = 159;
   tsAsyncLog = 0;
 
   char temp[128] = {0};
@@ -128,7 +132,7 @@ void ctgTestBuildCTableMetaOutput(STableMetaOutput *output) {
   char tbFullName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(&cn, tbFullName);
 
-  output->metaNum = 2;
+  SET_META_TYPE_BOTH_TABLE(output->metaType);
 
   strcpy(output->ctbFname, tbFullName);
 
@@ -183,6 +187,7 @@ void ctgTestBuildDBVgroup(SDBVgroupInfo *dbVgroup) {
   ctgTestCurrentVgVersion = dbVgroup->vgVersion;
   
   dbVgroup->hashMethod = 0;
+  dbVgroup->dbId = ctgTestDbId;
   dbVgroup->vgInfo = taosHashInit(ctgTestVgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
 
   vgNum = ctgTestGetVgNumFromVgVersion(dbVgroup->vgVersion);
@@ -216,6 +221,7 @@ void ctgTestPrepareDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcM
   ctgTestCurrentVgVersion = ctgTestVgVersion;
   rspMsg->vgNum = htonl(ctgTestVgNum);
   rspMsg->hashMethod = 0;
+  rspMsg->uid = htobe64(ctgTestDbId);
 
   SVgroupInfo *vg = NULL;
   uint32_t hashUnit = UINT32_MAX / ctgTestVgNum;
@@ -338,8 +344,52 @@ void ctgTestPrepareSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
   rspMsg->update = 1;
   rspMsg->sversion = htonl(ctgTestSVersion);
   rspMsg->tversion = htonl(ctgTestTVersion);
-  rspMsg->suid = htobe64(0x0000000000000002);
-  rspMsg->tuid = htobe64(0x0000000000000003);
+  rspMsg->suid = htobe64(ctgTestSuid);
+  rspMsg->tuid = htobe64(ctgTestSuid);
+  rspMsg->vgId = 0;
+
+  SSchema *s = NULL;
+  s = &rspMsg->pSchema[0];
+  s->type = TSDB_DATA_TYPE_TIMESTAMP;
+  s->colId = htonl(1);
+  s->bytes = htonl(8);
+  strcpy(s->name, "ts");
+
+  s = &rspMsg->pSchema[1];
+  s->type = TSDB_DATA_TYPE_INT;
+  s->colId = htonl(2);
+  s->bytes = htonl(4);
+  strcpy(s->name, "col1s");
+
+  s = &rspMsg->pSchema[2];
+  s->type = TSDB_DATA_TYPE_BINARY;
+  s->colId = htonl(3);
+  s->bytes = htonl(12);
+  strcpy(s->name, "tag1s");
+
+  
+  return;
+}
+
+void ctgTestPrepareMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
+  STableMetaMsg *rspMsg = NULL; //todo
+  static int32_t idx = 1;
+
+  pRsp->code =0;
+  pRsp->contLen = sizeof(STableMetaMsg) + (ctgTestColNum + ctgTestTagNum) * sizeof(SSchema);
+  pRsp->pCont = calloc(1, pRsp->contLen);
+  rspMsg = (STableMetaMsg *)pRsp->pCont;
+  sprintf(rspMsg->tbFname, "%s.%s_%d", ctgTestDbname, ctgTestSTablename, idx);
+  sprintf(rspMsg->stbFname, "%s.%s_%d", ctgTestDbname, ctgTestSTablename, idx);
+  rspMsg->numOfTags = htonl(ctgTestTagNum);
+  rspMsg->numOfColumns = htonl(ctgTestColNum);
+  rspMsg->precision = 1;
+  rspMsg->tableType = TSDB_SUPER_TABLE;
+  rspMsg->update = 1;
+  rspMsg->sversion = htonl(ctgTestSVersion);
+  rspMsg->tversion = htonl(ctgTestTVersion);
+  rspMsg->suid = htobe64(ctgTestSuid + idx);
+  rspMsg->tuid = htobe64(ctgTestSuid + idx);
   rspMsg->vgId = 0;
 
   SSchema *s = NULL;
@@ -361,10 +411,13 @@ void ctgTestPrepareSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
   s->bytes = htonl(12);
   strcpy(s->name, "tag1s");
 
+  ++idx;
   
   return;
 }
 
+
+
 void ctgTestPrepareDbVgroupsAndNormalMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
   ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
   
@@ -390,6 +443,14 @@ void ctgTestPrepareDbVgroupsAndSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg
   return;
 }
 
+void ctgTestPrepareDbVgroupsAndMultiSuperMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) {
+  ctgTestPrepareDbVgroups(shandle, pEpSet, pMsg, pRsp);
+  
+  ctgTestSetPrepareMultiSTableMeta();
+  
+  return;
+}
+
 
 
 void ctgTestSetPrepareDbVgroups() {
@@ -444,6 +505,20 @@ void ctgTestSetPrepareSTableMeta() {
   }
 }
 
+void ctgTestSetPrepareMultiSTableMeta() {
+  static Stub stub;
+  stub.set(rpcSendRecv, ctgTestPrepareMultiSTableMeta);
+  {
+    AddrAny any("libtransport.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, ctgTestPrepareMultiSTableMeta);
+    }
+  }
+}
+
+
 void ctgTestSetPrepareDbVgroupsAndNormalMeta() {
   static Stub stub;
   stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndNormalMeta);
@@ -484,6 +559,19 @@ void ctgTestSetPrepareDbVgroupsAndSuperMeta() {
   }
 }
 
+void ctgTestSetPrepareDbVgroupsAndMultiSuperMeta() {
+  static Stub stub;
+  stub.set(rpcSendRecv, ctgTestPrepareDbVgroupsAndMultiSuperMeta);
+  {
+    AddrAny any("libtransport.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^rpcSendRecv$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, ctgTestPrepareDbVgroupsAndMultiSuperMeta);
+    }
+  }
+}
+
 
 }
 
@@ -507,7 +595,7 @@ void *ctgTestGetDbVgroupThread(void *param) {
     if (ctgTestEnableSleep) {
       usleep(rand()%5);
     }
-    if (++n % 50000 == 0) {
+    if (++n % ctgTestPrintNum == 0) {
       printf("Get:%d\n", n);
     }
   }
@@ -531,7 +619,7 @@ void *ctgTestSetDbVgroupThread(void *param) {
     if (ctgTestEnableSleep) {    
       usleep(rand()%5);
     }
-    if (++n % 50000 == 0) {
+    if (++n % ctgTestPrintNum == 0) {
       printf("Set:%d\n", n);
     }
   }
@@ -563,7 +651,7 @@ void *ctgTestGetCtableMetaThread(void *param) {
       usleep(rand()%5);
     }
     
-    if (++n % 50000 == 0) {
+    if (++n % ctgTestPrintNum == 0) {
       printf("Get:%d\n", n);
     }
   }
@@ -589,7 +677,7 @@ void *ctgTestSetCtableMetaThread(void *param) {
     if (ctgTestEnableSleep) {    
       usleep(rand()%5);
     }
-    if (++n % 50000 == 0) {
+    if (++n % ctgTestPrintNum == 0) {
       printf("Set:%d\n", n);
     }
   }
@@ -600,7 +688,6 @@ void *ctgTestSetCtableMetaThread(void *param) {
 
 }
 
-#if 0
 
 TEST(tableMeta, normalTable) {
   struct SCatalog* pCtg = NULL;
@@ -628,6 +715,7 @@ TEST(tableMeta, normalTable) {
   ASSERT_EQ(vgInfo.vgId, 8);
   ASSERT_EQ(vgInfo.numOfEps, 3);
 
+
   ctgTestSetPrepareTableMeta();
 
   STableMeta *tableMeta = NULL;
@@ -654,6 +742,41 @@ TEST(tableMeta, normalTable) {
   ASSERT_EQ(tableMeta->tableInfo.precision, 1);
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
+  SDbVgVersion *dbs = NULL;
+  SSTableMetaVersion *stb = NULL;
+  uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
+  int32_t i = 0;
+  while (i < 5) {
+    ++i;
+    code = catalogGetExpiredDBs(pCtg, &dbs, &dbNum);
+    ASSERT_EQ(code, 0);
+    code = catalogGetExpiredSTables(pCtg, &stb, &stbNum);
+    ASSERT_EQ(code, 0);
+    
+    if (dbNum) {
+      printf("got expired db,dbId:%"PRId64"\n", dbs->dbId);
+      free(dbs);
+      dbs = NULL;
+    } else {
+      printf("no expired db\n");
+    }
+
+    if (stbNum) {
+      printf("got expired stb,suid:%"PRId64"\n", stb->suid);
+      free(stb);
+      stb = NULL;
+    } else {
+      printf("no expired stb\n");
+    }
+
+    allDbNum += dbNum;
+    allStbNum += stbNum;
+    sleep(2);
+  }
+  
+  ASSERT_EQ(allDbNum, 1);
+  ASSERT_EQ(allStbNum, 0);
+
   catalogDestroy();
 }
 
@@ -715,6 +838,42 @@ TEST(tableMeta, childTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.precision, 1);
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
+  SDbVgVersion *dbs = NULL;
+  SSTableMetaVersion *stb = NULL;
+  uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
+  int32_t i = 0;
+  while (i < 5) {
+    ++i;
+    code = catalogGetExpiredDBs(pCtg, &dbs, &dbNum);
+    ASSERT_EQ(code, 0);
+    code = catalogGetExpiredSTables(pCtg, &stb, &stbNum);
+    ASSERT_EQ(code, 0);
+    
+    if (dbNum) {
+      printf("got expired db,dbId:%"PRId64"\n", dbs->dbId);
+      free(dbs);
+      dbs = NULL;
+    } else {
+      printf("no expired db\n");
+    }
+
+    if (stbNum) {
+      printf("got expired stb,suid:%"PRId64"\n", stb->suid);
+      free(stb);
+      stb = NULL;
+    } else {
+      printf("no expired stb\n");
+    }
+
+    allDbNum += dbNum;
+    allStbNum += stbNum;
+    sleep(2);
+  }
+  
+  ASSERT_EQ(allDbNum, 1);
+  ASSERT_EQ(allStbNum, 1);
+
+
   catalogDestroy();
 }
 
@@ -745,6 +904,8 @@ TEST(tableMeta, superTableCase) {
   ASSERT_EQ(tableMeta->tableType, TSDB_SUPER_TABLE);
   ASSERT_EQ(tableMeta->sversion, ctgTestSVersion);
   ASSERT_EQ(tableMeta->tversion, ctgTestTVersion);
+  ASSERT_EQ(tableMeta->uid, ctgTestSuid);
+  ASSERT_EQ(tableMeta->suid, ctgTestSuid);
   ASSERT_EQ(tableMeta->tableInfo.numOfColumns, ctgTestColNum);
   ASSERT_EQ(tableMeta->tableInfo.numOfTags, ctgTestTagNum);
   ASSERT_EQ(tableMeta->tableInfo.precision, 1);
@@ -768,7 +929,7 @@ TEST(tableMeta, superTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
   tableMeta = NULL;
-  code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
+  code = catalogRenewAndGetTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta, 0);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(tableMeta->vgId, 9);
   ASSERT_EQ(tableMeta->tableType, TSDB_CHILD_TABLE);
@@ -779,6 +940,40 @@ TEST(tableMeta, superTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.precision, 1);
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
+  SDbVgVersion *dbs = NULL;
+  SSTableMetaVersion *stb = NULL;
+  uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
+  int32_t i = 0;
+  while (i < 5) {
+    ++i;
+    code = catalogGetExpiredDBs(pCtg, &dbs, &dbNum);
+    ASSERT_EQ(code, 0);
+    code = catalogGetExpiredSTables(pCtg, &stb, &stbNum);
+    ASSERT_EQ(code, 0);
+    
+    if (dbNum) {
+      printf("got expired db,dbId:%"PRId64"\n", dbs->dbId);
+      free(dbs);
+      dbs = NULL;
+    } else {
+      printf("no expired db\n");
+    }
+
+    if (stbNum) {
+      printf("got expired stb,suid:%"PRId64"\n", stb->suid);
+      free(stb);
+      stb = NULL;
+    } else {
+      printf("no expired stb\n");
+    }
+
+    allDbNum += dbNum;
+    allStbNum += stbNum;
+    sleep(2);
+  }
+  
+  ASSERT_EQ(allDbNum, 1);
+  ASSERT_EQ(allStbNum, 1);
 
 
   catalogDestroy();
@@ -948,7 +1143,6 @@ TEST(dbVgroup, getSetDbVgroupCase) {
   catalogDestroy();
 }
 
-
 TEST(multiThread, getSetDbVgroupCase) {
   struct SCatalog* pCtg = NULL;
   void *mockPointer = (void *)0x1;
@@ -956,6 +1150,7 @@ TEST(multiThread, getSetDbVgroupCase) {
   SVgroupInfo *pvgInfo = NULL;
   SDBVgroupInfo dbVgroup = {0};
   SArray *vgList = NULL;
+  ctgTestStop = false;
 
   ctgTestInitLogFile();
   
@@ -988,7 +1183,7 @@ TEST(multiThread, getSetDbVgroupCase) {
     if (ctgTestDeadLoop) {
       sleep(1);
     } else {
-      sleep(600);
+      sleep(ctgTestMTRunSec);
       break;
     }
   }
@@ -999,9 +1194,6 @@ TEST(multiThread, getSetDbVgroupCase) {
   catalogDestroy();
 }
 
-#endif
-
-
 TEST(multiThread, ctableMeta) {
   struct SCatalog* pCtg = NULL;
   void *mockPointer = (void *)0x1;
@@ -1009,6 +1201,7 @@ TEST(multiThread, ctableMeta) {
   SVgroupInfo *pvgInfo = NULL;
   SDBVgroupInfo dbVgroup = {0};
   SArray *vgList = NULL;
+  ctgTestStop = false;
 
   ctgTestSetPrepareDbVgroupsAndChildMeta();
 
@@ -1038,7 +1231,7 @@ TEST(multiThread, ctableMeta) {
     if (ctgTestDeadLoop) {
       sleep(1);
     } else {
-      sleep(600);
+      sleep(ctgTestMTRunSec);
       break;
     }
   }
@@ -1050,6 +1243,78 @@ TEST(multiThread, ctableMeta) {
 }
 
 
+TEST(rentTest, allRent) {
+  struct SCatalog* pCtg = NULL;
+  void *mockPointer = (void *)0x1;
+  SVgroupInfo vgInfo = {0};
+  SVgroupInfo *pvgInfo = NULL;
+  SDBVgroupInfo dbVgroup = {0};
+  SArray *vgList = NULL;
+  ctgTestStop = false;
+  SDbVgVersion *dbs = NULL;
+  SSTableMetaVersion *stable = NULL;
+  uint32_t num = 0;
+
+  ctgTestSetPrepareDbVgroupsAndMultiSuperMeta();
+
+  initQueryModuleMsgHandle();
+
+  int32_t code = catalogInit(NULL);
+  ASSERT_EQ(code, 0);
+
+  code = catalogGetHandle(ctgTestClusterId, &pCtg);
+  ASSERT_EQ(code, 0);
+
+
+  SName n = {.type = TSDB_TABLE_NAME_T, .acctId = 1};
+  strcpy(n.dbname, "db1");
+
+  for (int32_t i = 1; i <= 10; ++i) {
+    sprintf(n.tname, "%s_%d", ctgTestSTablename, i);
+
+    STableMeta *tableMeta = NULL;
+    code = catalogGetSTableMeta(pCtg, mockPointer, (const SEpSet *)mockPointer, &n, &tableMeta);
+    ASSERT_EQ(code, 0);
+    ASSERT_EQ(tableMeta->vgId, 0);
+    ASSERT_EQ(tableMeta->tableType, TSDB_SUPER_TABLE);
+    ASSERT_EQ(tableMeta->sversion, ctgTestSVersion);
+    ASSERT_EQ(tableMeta->tversion, ctgTestTVersion);
+    ASSERT_EQ(tableMeta->uid, ctgTestSuid + i);
+    ASSERT_EQ(tableMeta->suid, ctgTestSuid + i);
+    ASSERT_EQ(tableMeta->tableInfo.numOfColumns, ctgTestColNum);
+    ASSERT_EQ(tableMeta->tableInfo.numOfTags, ctgTestTagNum);
+    ASSERT_EQ(tableMeta->tableInfo.precision, 1);
+    ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
+    
+    code = catalogGetExpiredDBs(pCtg, &dbs, &num);
+    ASSERT_EQ(code, 0);
+    printf("%d - expired dbNum:%d\n", i, num);
+    if (dbs) {
+      printf("%d - expired dbId:%"PRId64", vgVersion:%d\n", i, dbs->dbId, dbs->vgVersion);
+      free(dbs);
+      dbs = NULL;
+    }
+    
+    code = catalogGetExpiredSTables(pCtg, &stable, &num);
+    ASSERT_EQ(code, 0);
+    printf("%d - expired stableNum:%d\n", i, num);
+    if (stable) {
+      for (int32_t n = 0; n < num; ++n) {
+        printf("suid:%"PRId64", sversion:%d, tversion:%d\n", stable[n].suid, stable[n].sversion, stable[n].tversion);
+      }
+      free(stable);
+      stable = NULL;
+    }
+    printf("*************************************************\n");
+    
+    sleep(2);
+  }
+  
+  catalogDestroy();
+}
+
+
+
 int main(int argc, char** argv) {
   testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();

source/libs/qcom/src/querymsg.c

@@ -97,6 +97,7 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
   
   pRsp->vgVersion = ntohl(pRsp->vgVersion);
   pRsp->vgNum = ntohl(pRsp->vgNum);
+  pRsp->uid = be64toh(pRsp->uid);
 
   if (pRsp->vgNum < 0) {
     qError("invalid db[%s] vgroup number[%d]", pRsp->db, pRsp->vgNum);
@@ -111,6 +112,7 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
 
   pOut->dbVgroup.vgVersion = pRsp->vgVersion;
   pOut->dbVgroup.hashMethod = pRsp->hashMethod;
+  pOut->dbVgroup.dbId = pRsp->uid;
   pOut->dbVgroup.vgInfo = taosHashInit(pRsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
   if (NULL == pOut->dbVgroup.vgInfo) {
     qError("hash init[%d] failed", pRsp->vgNum);
@@ -149,8 +151,8 @@ static int32_t queryConvertTableMetaMsg(STableMetaMsg* pMetaMsg) {
   pMetaMsg->numOfColumns = ntohl(pMetaMsg->numOfColumns);
   pMetaMsg->sversion = ntohl(pMetaMsg->sversion);
   pMetaMsg->tversion = ntohl(pMetaMsg->tversion);
-  pMetaMsg->tuid = htobe64(pMetaMsg->tuid);
-  pMetaMsg->suid = htobe64(pMetaMsg->suid);
+  pMetaMsg->tuid = be64toh(pMetaMsg->tuid);
+  pMetaMsg->suid = be64toh(pMetaMsg->suid);
   pMetaMsg->vgId = ntohl(pMetaMsg->vgId);
 
   if (pMetaMsg->numOfTags < 0 || pMetaMsg->numOfTags > TSDB_MAX_TAGS) {
@@ -208,7 +210,7 @@ int32_t queryCreateTableMetaFromMsg(STableMetaMsg* msg, bool isSuperTable, STabl
 
   pTableMeta->vgId = isSuperTable ? 0 : msg->vgId;
   pTableMeta->tableType = isSuperTable ? TSDB_SUPER_TABLE : msg->tableType;
-  pTableMeta->uid  = msg->tuid;
+  pTableMeta->uid  = isSuperTable ? msg->suid : msg->tuid;
   pTableMeta->suid = msg->suid;
   pTableMeta->sversion = msg->sversion;
   pTableMeta->tversion = msg->tversion;
@@ -244,7 +246,7 @@ int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
   }
 
   if (pMetaMsg->tableType == TSDB_CHILD_TABLE) {
-    pOut->metaNum = 2;
+    SET_META_TYPE_BOTH_TABLE(pOut->metaType);
 
     if (pMetaMsg->dbFname[0]) {
       snprintf(pOut->ctbFname, sizeof(pOut->ctbFname), "%s.%s", pMetaMsg->dbFname, pMetaMsg->tbFname);
@@ -261,7 +263,7 @@ int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
 
     code = queryCreateTableMetaFromMsg(pMetaMsg, true, &pOut->tbMeta);
   } else {
-    pOut->metaNum = 1;
+    SET_META_TYPE_TABLE(pOut->metaType);
     
     if (pMetaMsg->dbFname[0]) {
       snprintf(pOut->tbFname, sizeof(pOut->tbFname), "%s.%s", pMetaMsg->dbFname, pMetaMsg->tbFname);

source/libs/scheduler/src/scheduler.c

@@ -505,7 +505,7 @@ int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *ms
       break;
     }
     case TDMT_VND_SUBMIT_RSP: {
-        if (rspCode != TSDB_CODE_SUCCESS) {
+        if (rspCode != TSDB_CODE_SUCCESS || NULL == msg) {
           SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rspCode));
         } else {
           SShellSubmitRspMsg *rsp = (SShellSubmitRspMsg *)msg;
@@ -521,7 +521,7 @@ int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *ms
     case TDMT_VND_QUERY_RSP: {
         SQueryTableRsp *rsp = (SQueryTableRsp *)msg;
         
-        if (rsp->code != TSDB_CODE_SUCCESS) {
+        if (rsp->code != TSDB_CODE_SUCCESS || NULL == msg) {
           SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rsp->code));
         } else {
           code = schBuildAndSendMsg(job, task, TDMT_VND_RES_READY);
@@ -534,7 +534,7 @@ int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *ms
     case TDMT_VND_RES_READY_RSP: {
         SResReadyRsp *rsp = (SResReadyRsp *)msg;
         
-        if (rsp->code != TSDB_CODE_SUCCESS) {
+        if (rsp->code != TSDB_CODE_SUCCESS || NULL == msg) {
           SCH_ERR_JRET(schProcessOnTaskFailure(job, task, rsp->code));
         } else {
           code = schProcessOnTaskSuccess(job, task);
@@ -549,7 +549,9 @@ int32_t schProcessRspMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *ms
         SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
 
         job->res = rsp;
+        if (rsp) {
           job->resNumOfRows = rsp->numOfRows;
+        }
         
         SCH_ERR_JRET(schProcessOnDataFetched(job));
         break;
@@ -1100,6 +1102,7 @@ void scheduleFreeJob(void *pJob) {
   taosHashCleanup(job->failTasks);
   taosHashCleanup(job->succTasks);
   taosArrayDestroy(job->levels);
+  
   tfree(job);
 }
 

source/util/src/thash.c

@@ -215,7 +215,7 @@ static FORCE_INLINE bool taosHashTableEmpty(const SHashObj *pHashObj) {
   return taosHashGetSize(pHashObj) == 0;
 }
 
-int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size) {
+int32_t taosHashPutImpl(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size, bool *newAdded) {
   uint32_t   hashVal = (*pHashObj->hashFp)(key, (uint32_t)keyLen);
   SHashNode *pNewNode = doCreateHashNode(key, keyLen, data, size, hashVal);
   if (pNewNode == NULL) {
@@ -274,6 +274,10 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
     __rd_unlock((void*) &pHashObj->lock, pHashObj->type);
     atomic_add_fetch_32(&pHashObj->size, 1);
 
+    if (newAdded) {
+      *newAdded = true;
+    }
+    
     return 0;
   } else {
     // not support the update operation, return error
@@ -290,10 +294,23 @@ int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *da
     // enable resize
     __rd_unlock((void*) &pHashObj->lock, pHashObj->type);
 
+    if (newAdded) {
+      *newAdded = false;
+    }
+
     return pHashObj->enableUpdate ? 0 : -2;
   }
 }
 
+int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size) {
+  return taosHashPutImpl(pHashObj, key, keyLen, data, size, NULL);
+}
+
+int32_t taosHashPutExt(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size, bool *newAdded) {
+  return taosHashPutImpl(pHashObj, key, keyLen, data, size, newAdded);
+}
+
+
 void *taosHashGet(SHashObj *pHashObj, const void *key, size_t keyLen) {
   return taosHashGetClone(pHashObj, key, keyLen, NULL);
 }

source/util/src/tlog.c

@@ -95,7 +95,6 @@ int32_t tsdbDebugFlag = 131;
 int32_t tqDebugFlag = 131;
 int32_t cqDebugFlag = 131;
 int32_t fsDebugFlag = 135;
-int32_t ctgDebugFlag = 131;
 
 int64_t dbgEmptyW = 0;
 int64_t dbgWN = 0;