Merge remote-tracking branch 'origin/3.0' into fix/mnode

include/common/tmsg.h

@@ -1134,14 +1134,16 @@ void    tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp);
 int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
 int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
 void    tFreeSTableMetaRsp(STableMetaRsp* pRsp);
+void tFreeSTableIndexRsp(void *info);
 
 typedef struct {
-  SArray* pArray;  // Array of STableMetaRsp
-} STableMetaBatchRsp;
+  SArray*         pMetaRsp;  // Array of STableMetaRsp
+  SArray*         pIndexRsp;  // Array of STableIndexRsp;
+} SSTbHbRsp;
 
-int32_t tSerializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
-int32_t tDeserializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
-void    tFreeSTableMetaBatchRsp(STableMetaBatchRsp* pRsp);
+int32_t tSerializeSSTbHbRsp(void* buf, int32_t bufLen, SSTbHbRsp* pRsp);
+int32_t tDeserializeSSTbHbRsp(void* buf, int32_t bufLen, SSTbHbRsp* pRsp);
+void    tFreeSSTbHbRsp(SSTbHbRsp* pRsp);
 
 typedef struct {
   int32_t numOfTables;
@@ -2502,7 +2504,11 @@ typedef struct {
 } STableIndexInfo;
 
 typedef struct {
-  SArray* pIndex;
+  char     tbName[TSDB_TABLE_NAME_LEN];
+  char     dbFName[TSDB_DB_FNAME_LEN];
+  uint64_t suid;
+  int32_t  version;
+  SArray*  pIndex;
 } STableIndexRsp;
 
 int32_t tSerializeSTableIndexRsp(void* buf, int32_t bufLen, const STableIndexRsp* pRsp);

include/libs/catalog/catalog.h

@@ -98,14 +98,15 @@ typedef struct SCatalogCfg {
   uint32_t stbRentSec;
 } SCatalogCfg;
 
-typedef struct SSTableMetaVersion {
+typedef struct SSTableVersion {
   char     dbFName[TSDB_DB_FNAME_LEN];
   char     stbName[TSDB_TABLE_NAME_LEN];
   uint64_t dbId;
   uint64_t suid;
   int16_t  sversion;
-  int16_t  tversion;
-} SSTableMetaVersion;
+  int16_t  tversion;  
+  int32_t  smaVer;
+} SSTableVersion;
 
 typedef struct SDbVgVersion {
   char    dbFName[TSDB_DB_FNAME_LEN];
@@ -267,7 +268,7 @@ int32_t catalogAsyncGetAllMeta(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t
 
 int32_t catalogGetQnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArray* pQnodeList);
 
-int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableMetaVersion** stables, uint32_t* num);
+int32_t catalogGetExpiredSTables(SCatalog* pCatalog, SSTableVersion **stables, uint32_t *num);
 
 int32_t catalogGetExpiredDBs(SCatalog* pCatalog, SDbVgVersion** dbs, uint32_t* num);
 
@@ -279,6 +280,8 @@ int32_t catalogGetIndexMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const char*
 
 int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes);
 
+int32_t catalogUpdateTableIndex(SCatalog* pCtg, STableIndexRsp *pRsp);
+
 int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* funcName, SFuncInfo* pInfo);
 
 int32_t catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass);

include/libs/sync/sync.h

@@ -47,8 +47,9 @@ typedef enum {
 typedef enum {
   TAOS_SYNC_PROPOSE_SUCCESS = 0,
   TAOS_SYNC_PROPOSE_NOT_LEADER = 1,
-  TAOS_SYNC_PROPOSE_OTHER_ERROR = 2,
-  TAOS_SYNC_ONLY_ONE_REPLICA = 3,
+  TAOS_SYNC_ONLY_ONE_REPLICA = 2,
+  TAOS_SYNC_NOT_IN_NEW_CONFIG = 3,
+  TAOS_SYNC_OTHER_ERROR = 100,
 } ESyncProposeCode;
 
 typedef enum {
@@ -110,6 +111,7 @@ typedef struct SSyncFSM {
 
   void (*FpRestoreFinishCb)(struct SSyncFSM* pFsm);
   void (*FpReConfigCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta);
+  void (*FpLeaderTransferCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
 
   int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot);
 
@@ -199,15 +201,13 @@ bool        syncIsRestoreFinish(int64_t rid);
 int32_t     syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
 
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg);
-int32_t syncReconfigRaw(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg);
+
+// build SRpcMsg, need to call syncPropose with SRpcMsg
+int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg);
 
 int32_t syncLeaderTransfer(int64_t rid);
 int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader);
 
-// to be moved to static
-void syncStartNormal(int64_t rid);
-void syncStartStandBy(int64_t rid);
-
 #ifdef __cplusplus
 }
 #endif

include/libs/sync/syncTools.h

@@ -467,6 +467,7 @@ typedef struct SyncLeaderTransfer {
    SRaftId  srcId;
    SRaftId  destId;
    */
+  SNodeInfo newNodeInfo;
   SRaftId newLeaderId;
 } SyncLeaderTransfer;
 

source/client/src/clientHb.c

@@ -99,15 +99,15 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
 static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
   int32_t code = 0;
 
-  STableMetaBatchRsp batchMetaRsp = {0};
-  if (tDeserializeSTableMetaBatchRsp(value, valueLen, &batchMetaRsp) != 0) {
+  SSTbHbRsp hbRsp = {0};
+  if (tDeserializeSSTbHbRsp(value, valueLen, &hbRsp) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
 
-  int32_t numOfBatchs = taosArrayGetSize(batchMetaRsp.pArray);
-  for (int32_t i = 0; i < numOfBatchs; ++i) {
-    STableMetaRsp *rsp = taosArrayGet(batchMetaRsp.pArray, i);
+  int32_t numOfMeta = taosArrayGetSize(hbRsp.pMetaRsp);
+  for (int32_t i = 0; i < numOfMeta; ++i) {
+    STableMetaRsp *rsp = taosArrayGet(hbRsp.pMetaRsp, i);
 
     if (rsp->numOfColumns < 0) {
       tscDebug("hb remove stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
@@ -116,7 +116,7 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
       tscDebug("hb update stb, db:%s, stb:%s", rsp->dbFName, rsp->stbName);
       if (rsp->pSchemas[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
         tscError("invalid colId[%" PRIi16 "] for the first column in table meta rsp msg", rsp->pSchemas[0].colId);
-        tFreeSTableMetaBatchRsp(&batchMetaRsp);
+        tFreeSSTbHbRsp(&hbRsp);
         return TSDB_CODE_TSC_INVALID_VALUE;
       }
 
@@ -124,7 +124,17 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
     }
   }
 
-  tFreeSTableMetaBatchRsp(&batchMetaRsp);
+  int32_t numOfIndex = taosArrayGetSize(hbRsp.pIndexRsp);
+  for (int32_t i = 0; i < numOfIndex; ++i) {
+    STableIndexRsp *rsp = taosArrayGet(hbRsp.pIndexRsp, i);
+
+    catalogUpdateTableIndex(pCatalog, rsp);
+  }
+
+  taosArrayDestroy(hbRsp.pIndexRsp);
+  hbRsp.pIndexRsp = NULL;
+
+  tFreeSSTbHbRsp(&hbRsp);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -455,7 +465,7 @@ int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SCl
 }
 
 int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
-  SSTableMetaVersion *stbs = NULL;
+  SSTableVersion *stbs = NULL;
   uint32_t            stbNum = 0;
   int32_t             code = 0;
 
@@ -469,15 +479,16 @@ int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SC
   }
 
   for (int32_t i = 0; i < stbNum; ++i) {
-    SSTableMetaVersion *stb = &stbs[i];
+    SSTableVersion *stb = &stbs[i];
     stb->suid = htobe64(stb->suid);
     stb->sversion = htons(stb->sversion);
     stb->tversion = htons(stb->tversion);
+    stb->smaVer = htonl(stb->smaVer);
   }
 
   SKv kv = {
       .key = HEARTBEAT_KEY_STBINFO,
-      .valueLen = sizeof(SSTableMetaVersion) * stbNum,
+      .valueLen = sizeof(SSTableVersion) * stbNum,
       .value = stbs,
   };
 

source/common/src/tmsg.c

@@ -2438,6 +2438,10 @@ int32_t tSerializeSTableIndexRsp(void *buf, int32_t bufLen, const STableIndexRsp
   tEncoderInit(&encoder, buf, bufLen);
 
   if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeCStr(&encoder, pRsp->tbName) < 0) return -1;
+  if (tEncodeCStr(&encoder, pRsp->dbFName) < 0) return -1;
+  if (tEncodeU64(&encoder, pRsp->suid) < 0) return -1;
+  if (tEncodeI32(&encoder, pRsp->version) < 0) return -1;
   int32_t num = taosArrayGetSize(pRsp->pIndex);
   if (tEncodeI32(&encoder, num) < 0) return -1;
   if (num > 0) {
@@ -2472,6 +2476,10 @@ int32_t tDeserializeSTableIndexRsp(void *buf, int32_t bufLen, STableIndexRsp *pR
   tDecoderInit(&decoder, buf, bufLen);
 
   if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pRsp->tbName) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pRsp->dbFName) < 0) return -1;
+  if (tDecodeU64(&decoder, &pRsp->suid) < 0) return -1;
+  if (tDecodeI32(&decoder, &pRsp->version) < 0) return -1;
   int32_t num = 0;
   if (tDecodeI32(&decoder, &num) < 0) return -1;
   if (num > 0) {
@@ -2631,18 +2639,35 @@ int32_t tSerializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp)
   return tlen;
 }
 
-int32_t tSerializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) {
+int32_t tSerializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
 
   if (tStartEncode(&encoder) < 0) return -1;
 
-  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
-  if (tEncodeI32(&encoder, numOfBatch) < 0) return -1;
-  for (int32_t i = 0; i < numOfBatch; ++i) {
-    STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i);
+  int32_t numOfMeta = taosArrayGetSize(pRsp->pMetaRsp);
+  if (tEncodeI32(&encoder, numOfMeta) < 0) return -1;
+  for (int32_t i = 0; i < numOfMeta; ++i) {
+    STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pMetaRsp, i);
     if (tEncodeSTableMetaRsp(&encoder, pMetaRsp) < 0) return -1;
   }
+
+  int32_t numOfIndex = taosArrayGetSize(pRsp->pIndexRsp);
+  if (tEncodeI32(&encoder, numOfIndex) < 0) return -1;  
+  for (int32_t i = 0; i < numOfIndex; ++i) {
+    STableIndexRsp *pIndexRsp = taosArrayGet(pRsp->pIndexRsp, i);
+    if (tEncodeCStr(&encoder, pIndexRsp->tbName) < 0) return -1;
+    if (tEncodeCStr(&encoder, pIndexRsp->dbFName) < 0) return -1;
+    if (tEncodeU64(&encoder, pIndexRsp->suid) < 0) return -1;
+    if (tEncodeI32(&encoder, pIndexRsp->version) < 0) return -1;
+    int32_t num = taosArrayGetSize(pIndexRsp->pIndex);
+    if (tEncodeI32(&encoder, num) < 0) return -1;
+    for (int32_t i = 0; i < num; ++i) {
+      STableIndexInfo *pInfo = (STableIndexInfo *)taosArrayGet(pIndexRsp->pIndex, i);
+      if (tSerializeSTableIndexInfo(&encoder, pInfo) < 0) return -1;
+    }
+  }
+
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -2662,26 +2687,58 @@ int32_t tDeserializeSTableMetaRsp(void *buf, int32_t bufLen, STableMetaRsp *pRsp
   return 0;
 }
 
-int32_t tDeserializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatchRsp *pRsp) {
+int32_t tDeserializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
   SDecoder decoder = {0};
   tDecoderInit(&decoder, buf, bufLen);
 
   if (tStartDecode(&decoder) < 0) return -1;
 
-  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
-  if (tDecodeI32(&decoder, &numOfBatch) < 0) return -1;
-
-  pRsp->pArray = taosArrayInit(numOfBatch, sizeof(STableMetaRsp));
-  if (pRsp->pArray == NULL) {
+  int32_t numOfMeta = 0;
+  if (tDecodeI32(&decoder, &numOfMeta) < 0) return -1;
+  pRsp->pMetaRsp = taosArrayInit(numOfMeta, sizeof(STableMetaRsp));
+  if (pRsp->pMetaRsp == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
-  for (int32_t i = 0; i < numOfBatch; ++i) {
+  for (int32_t i = 0; i < numOfMeta; ++i) {
     STableMetaRsp tableMetaRsp = {0};
     if (tDecodeSTableMetaRsp(&decoder, &tableMetaRsp) < 0) return -1;
-    taosArrayPush(pRsp->pArray, &tableMetaRsp);
+    taosArrayPush(pRsp->pMetaRsp, &tableMetaRsp);
   }
+
+  int32_t numOfIndex = 0;
+  if (tDecodeI32(&decoder, &numOfIndex) < 0) return -1;
+
+  pRsp->pIndexRsp = taosArrayInit(numOfIndex, sizeof(STableIndexRsp));
+  if (pRsp->pIndexRsp == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  for (int32_t i = 0; i < numOfIndex; ++i) {
+    STableIndexRsp tableIndexRsp = {0};
+    if (tDecodeCStrTo(&decoder, tableIndexRsp.tbName) < 0) return -1;
+    if (tDecodeCStrTo(&decoder, tableIndexRsp.dbFName) < 0) return -1;
+    if (tDecodeU64(&decoder, &tableIndexRsp.suid) < 0) return -1;
+    if (tDecodeI32(&decoder, &tableIndexRsp.version) < 0) return -1;
+    int32_t num = 0;
+    if (tDecodeI32(&decoder, &num) < 0) return -1;
+    if (num > 0) {
+      tableIndexRsp.pIndex = taosArrayInit(num, sizeof(STableIndexInfo));
+      if (NULL == tableIndexRsp.pIndex) return -1;
+      STableIndexInfo info;
+      for (int32_t i = 0; i < num; ++i) {
+        if (tDeserializeSTableIndexInfo(&decoder, &info) < 0) return -1;
+        if (NULL == taosArrayPush(tableIndexRsp.pIndex, &info)) {
+          taosMemoryFree(info.expr);
+          return -1;
+        }
+      }
+    }
+    taosArrayPush(pRsp->pIndexRsp, &tableIndexRsp);
+  }
+  
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);
@@ -2690,14 +2747,32 @@ int32_t tDeserializeSTableMetaBatchRsp(void *buf, int32_t bufLen, STableMetaBatc
 
 void tFreeSTableMetaRsp(STableMetaRsp *pRsp) { taosMemoryFreeClear(pRsp->pSchemas); }
 
-void tFreeSTableMetaBatchRsp(STableMetaBatchRsp *pRsp) {
-  int32_t numOfBatch = taosArrayGetSize(pRsp->pArray);
-  for (int32_t i = 0; i < numOfBatch; ++i) {
-    STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pArray, i);
+void tFreeSTableIndexRsp(void *info) {
+  if (NULL == info) {
+    return;
+  }
+
+  STableIndexRsp *pInfo = (STableIndexRsp *)info;
+
+  taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo);
+}
+
+void tFreeSSTbHbRsp(SSTbHbRsp *pRsp) {
+  int32_t numOfMeta = taosArrayGetSize(pRsp->pMetaRsp);
+  for (int32_t i = 0; i < numOfMeta; ++i) {
+    STableMetaRsp *pMetaRsp = taosArrayGet(pRsp->pMetaRsp, i);
     tFreeSTableMetaRsp(pMetaRsp);
   }
 
-  taosArrayDestroy(pRsp->pArray);
+  taosArrayDestroy(pRsp->pMetaRsp);
+
+  int32_t numOfIndex = taosArrayGetSize(pRsp->pIndexRsp);
+  for (int32_t i = 0; i < numOfIndex; ++i) {
+    STableIndexRsp *pIndexRsp = taosArrayGet(pRsp->pIndexRsp, i);
+    tFreeSTableIndexRsp(pIndexRsp);
+  }
+
+  taosArrayDestroy(pRsp->pIndexRsp);
 }
 
 int32_t tSerializeSShowRsp(void *buf, int32_t bufLen, SShowRsp *pRsp) {

source/dnode/mnode/impl/inc/mndSma.h

@@ -28,6 +28,7 @@ SSmaObj *mndAcquireSma(SMnode *pMnode, char *smaName);
 void     mndReleaseSma(SMnode *pMnode, SSmaObj *pSma);
 int32_t  mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb);
 int32_t  mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
+int32_t  mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist);
 
 #ifdef __cplusplus
 }

source/dnode/mnode/impl/inc/mndStb.h

@@ -27,7 +27,7 @@ void     mndCleanupStb(SMnode *pMnode);
 SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName);
 void     mndReleaseStb(SMnode *pMnode, SStbObj *pStb);
 SSdbRaw *mndStbActionEncode(SStbObj *pStb);
-int32_t  mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbs, int32_t numOfStbs, void **ppRsp,
+int32_t  mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbs, int32_t numOfStbs, void **ppRsp,
                             int32_t *pRspLen);
 int32_t  mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs);
 

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

@@ -380,22 +380,22 @@ void mndStop(SMnode *pMnode) {
 int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
   SMnode    *pMnode = pMsg->info.node;
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
-  int32_t    code = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+  int32_t    code = TAOS_SYNC_OTHER_ERROR;
 
   if (!syncEnvIsStart()) {
     mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
-    return TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    return TAOS_SYNC_OTHER_ERROR;
   }
 
   SSyncNode *pSyncNode = syncNodeAcquire(pMgmt->sync);
   if (pSyncNode == NULL) {
     mError("failed to process sync msg:%p type:%s since syncNode is null", pMsg, TMSG_INFO(pMsg->msgType));
-    return TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    return TAOS_SYNC_OTHER_ERROR;
   }
 
   if (mndAcquireSyncRef(pMnode) != 0) {
     mError("failed to process sync msg:%p type:%s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr());
-    return TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    return TAOS_SYNC_OTHER_ERROR;
   }
 
   char  logBuf[512] = {0};
@@ -456,7 +456,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       tmsgSendRsp(&rsp);
     } else {
       mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-      code = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+      code = TAOS_SYNC_OTHER_ERROR;
     }
   } else {
     if (pMsg->msgType == TDMT_SYNC_TIMEOUT) {
@@ -497,7 +497,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       tmsgSendRsp(&rsp);
     } else {
       mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-      code = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+      code = TAOS_SYNC_OTHER_ERROR;
     }
   }
 

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

@@ -433,7 +433,7 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
       case HEARTBEAT_KEY_STBINFO: {
         void   *rspMsg = NULL;
         int32_t rspLen = 0;
-        mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableMetaVersion), &rspMsg, &rspLen);
+        mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableVersion), &rspMsg, &rspLen);
         if (rspMsg && rspLen > 0) {
           SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
           taosArrayPush(hbRsp.info, &kv1);

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

@@ -552,6 +552,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
   SStreamObj streamObj = {0};
   tstrncpy(streamObj.name, pCreate->name, TSDB_STREAM_FNAME_LEN);
   tstrncpy(streamObj.sourceDb, pDb->name, TSDB_DB_FNAME_LEN);
+  tstrncpy(streamObj.targetDb, streamObj.sourceDb, TSDB_DB_FNAME_LEN);
   streamObj.createTime = taosGetTimestampMs();
   streamObj.updateTime = streamObj.createTime;
   streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
@@ -993,18 +994,31 @@ static int32_t mndGetSma(SMnode *pMnode, SUserIndexReq *indexReq, SUserIndexRsp
   return code;
 }
 
-static int32_t mndGetTableSma(SMnode *pMnode, STableIndexReq *indexReq, STableIndexRsp *rsp, bool *exist) {
+int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist) {
   int32_t         code = 0;
   SSmaObj        *pSma = NULL;
   SSdb           *pSdb = pMnode->pSdb;
   void           *pIter = NULL;
   STableIndexInfo info;
 
+  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
+  if (NULL == pStb) {
+    *exist = false;
+    return TSDB_CODE_SUCCESS;
+  }
+
+  strcpy(rsp->dbFName, pStb->db);
+  strcpy(rsp->tbName, pStb->name + strlen(pStb->db) + 1);
+  rsp->suid = pStb->uid;
+  rsp->version = pStb->smaVer;
+  mndReleaseStb(pMnode, pStb);
+  
+
   while (1) {
     pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
     if (pIter == NULL) break;
 
-    if (pSma->stb[0] != indexReq->tbFName[0] || strcmp(pSma->stb, indexReq->tbFName)) {
+    if (pSma->stb[0] != tbFName[0] || strcmp(pSma->stb, tbFName)) {
       continue;
     }
 
@@ -1116,7 +1130,7 @@ static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  code = mndGetTableSma(pMnode, &indexReq, &rsp, &exist);
+  code = mndGetTableSma(pMnode, indexReq.tbFName, &rsp, &exist);
   if (code) {
     goto _OVER;
   }
@@ -1207,4 +1221,4 @@ static int32_t mndRetrieveSma(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
 static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
-}
+}
\ No newline at end of file

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

@@ -28,6 +28,7 @@
 #include "mndTrans.h"
 #include "mndUser.h"
 #include "mndVgroup.h"
+#include "mndSma.h"
 #include "tname.h"
 
 #define STB_VER_NUMBER   1
@@ -1273,7 +1274,7 @@ static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbNa
   return 0;
 }
 
-static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
+static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp, int32_t *smaVer) {
   char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
   snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
 
@@ -1290,6 +1291,10 @@ static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char
     return -1;
   }
 
+  if (smaVer) {
+    *smaVer = pStb->smaVer;
+  }
+
   int32_t code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);
   mndReleaseDb(pMnode, pDb);
   mndReleaseStb(pMnode, pStb);
@@ -1637,7 +1642,7 @@ static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
     }
   } else {
     mDebug("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
-    if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) {
+    if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp, NULL) != 0) {
       goto _OVER;
     }
   }
@@ -1670,51 +1675,86 @@ _OVER:
   return code;
 }
 
-int32_t mndValidateStbInfo(SMnode *pMnode, SSTableMetaVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
+int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
                            int32_t *pRspLen) {
-  STableMetaBatchRsp batchMetaRsp = {0};
-  batchMetaRsp.pArray = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
-  if (batchMetaRsp.pArray == NULL) {
+  SSTbHbRsp hbRsp = {0};
+  hbRsp.pMetaRsp = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
+  if (hbRsp.pMetaRsp == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
+  hbRsp.pIndexRsp = taosArrayInit(numOfStbs, sizeof(STableIndexRsp));
+  if (NULL == hbRsp.pIndexRsp) {
+    taosArrayDestroy(hbRsp.pMetaRsp);
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+  
   for (int32_t i = 0; i < numOfStbs; ++i) {
-    SSTableMetaVersion *pStbVersion = &pStbVersions[i];
+    SSTableVersion *pStbVersion = &pStbVersions[i];
     pStbVersion->suid = be64toh(pStbVersion->suid);
     pStbVersion->sversion = ntohs(pStbVersion->sversion);
     pStbVersion->tversion = ntohs(pStbVersion->tversion);
+    pStbVersion->smaVer = ntohl(pStbVersion->smaVer);
 
     STableMetaRsp metaRsp = {0};
+    int32_t smaVer = 0;
     mDebug("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
-    if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) {
+    if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp, &smaVer) != 0) {
       metaRsp.numOfColumns = -1;
       metaRsp.suid = pStbVersion->suid;
+      taosArrayPush(hbRsp.pMetaRsp, &metaRsp);
+      continue;
     }
 
     if (pStbVersion->sversion != metaRsp.sversion || pStbVersion->tversion != metaRsp.tversion) {
-      taosArrayPush(batchMetaRsp.pArray, &metaRsp);
+      taosArrayPush(hbRsp.pMetaRsp, &metaRsp);
     } else {
       tFreeSTableMetaRsp(&metaRsp);
     }
+
+    if (pStbVersion->smaVer && pStbVersion->smaVer != smaVer) {
+      bool exist = false;
+      char tbFName[TSDB_TABLE_FNAME_LEN];
+      STableIndexRsp indexRsp = {0};
+      indexRsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));
+      if (NULL == indexRsp.pIndex) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
+      }
+      
+      sprintf(tbFName, "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);
+      int32_t code = mndGetTableSma(pMnode, tbFName, &indexRsp, &exist);
+      if (code || !exist) {
+        indexRsp.suid = pStbVersion->suid;
+        indexRsp.version = -1;
+        indexRsp.pIndex = NULL;
+      }
+
+      strcpy(indexRsp.dbFName, pStbVersion->dbFName);
+      strcpy(indexRsp.tbName, pStbVersion->stbName);
+
+      taosArrayPush(hbRsp.pIndexRsp, &indexRsp);
+    }
   }
 
-  int32_t rspLen = tSerializeSTableMetaBatchRsp(NULL, 0, &batchMetaRsp);
+  int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);
   if (rspLen < 0) {
-    tFreeSTableMetaBatchRsp(&batchMetaRsp);
+    tFreeSSTbHbRsp(&hbRsp);
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
 
   void *pRsp = taosMemoryMalloc(rspLen);
   if (pRsp == NULL) {
-    tFreeSTableMetaBatchRsp(&batchMetaRsp);
+    tFreeSSTbHbRsp(&hbRsp);
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
-  tSerializeSTableMetaBatchRsp(pRsp, rspLen, &batchMetaRsp);
-  tFreeSTableMetaBatchRsp(&batchMetaRsp);
+  tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
+  tFreeSSTbHbRsp(&hbRsp);
   *ppRsp = pRsp;
   *pRspLen = rspLen;
   return 0;

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

@@ -236,7 +236,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
     tsem_wait(&pMgmt->syncSem);
   } else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
     terrno = TSDB_CODE_APP_NOT_READY;
-  } else if (code == TAOS_SYNC_PROPOSE_OTHER_ERROR) {
+  } else if (code == TAOS_SYNC_OTHER_ERROR) {
     terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
   } else {
     terrno = TSDB_CODE_APP_ERROR;
@@ -254,13 +254,16 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
 void mndSyncStart(SMnode *pMnode) {
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
   syncSetMsgCb(pMgmt->sync, &pMnode->msgCb);
+  syncStart(pMgmt->sync);
+  mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby);
 
+/*
   if (pMgmt->standby) {
     syncStartStandBy(pMgmt->sync);
   } else {
     syncStart(pMgmt->sync);
   }
-  mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby);
+*/  
 }
 
 void mndSyncStop(SMnode *pMnode) {}

source/dnode/vnode/src/meta/metaQuery.c

@@ -663,12 +663,23 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
 
   void *  entryKey = NULL, *entryVal = NULL;
   int32_t nEntryKey, nEntryVal;
+  bool    first = true;
   while (1) {
     valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal);
     if (valid < 0) {
       break;
     }
     STagIdxKey *p = entryKey;
+    if (p->type != pCursor->type) {
+      if (first) {
+        valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);
+        if (valid < 0) break;
+        continue;
+      } else {
+        break;
+      }
+    }
+    first = false;
     if (p != NULL) {
       int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);
       if (cmp == 0) {

source/dnode/vnode/src/sma/smaTimeRange.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "sma.h"
+#include "tsdb.h"
+
+#define SMA_STORAGE_MINUTES_MAX  86400
+#define SMA_STORAGE_MINUTES_DAY  1440
+#define SMA_STORAGE_SPLIT_FACTOR 14400  // least records in tsma file
+
+/**
+ * @brief Judge the tsma file split days
+ *
+ * @param pCfg
+ * @param pCont
+ * @param contLen
+ * @param days unit is minute
+ * @return int32_t
+ */
+int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
+  SDecoder coder = {0};
+  tDecoderInit(&coder, pCont, contLen);
+
+  STSma tsma = {0};
+  if (tDecodeSVCreateTSmaReq(&coder, &tsma) < 0) {
+    terrno = TSDB_CODE_MSG_DECODE_ERROR;
+    goto _err;
+  }
+  STsdbCfg *pTsdbCfg = &pCfg->tsdbCfg;
+  int64_t   sInterval = convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_SECOND);
+  if (sInterval <= 0) {
+    *days = pTsdbCfg->days;
+    return 0;
+  }
+  int64_t records = pTsdbCfg->days * 60 / sInterval;
+  if (records >= SMA_STORAGE_SPLIT_FACTOR) {
+    *days = pTsdbCfg->days;
+  } else {
+    int64_t mInterval = convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_MINUTE);
+    int64_t daysPerFile = mInterval * SMA_STORAGE_MINUTES_DAY * 2;
+
+    if (daysPerFile > SMA_STORAGE_MINUTES_MAX) {
+      *days = SMA_STORAGE_MINUTES_MAX;
+    } else {
+      *days = (int32_t)daysPerFile;
+    }
+
+    if (*days < pTsdbCfg->days) {
+      *days = pTsdbCfg->days;
+    }
+  }
+  tDecoderClear(&coder);
+  return 0;
+_err:
+  tDecoderClear(&coder);
+  return -1;
+}
+
+/**
+ * @brief create tsma meta and result stable
+ *
+ * @param pSma
+ * @param version
+ * @param pMsg
+ * @return int32_t
+ */
+int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
+  SSmaCfg *pCfg = (SSmaCfg *)pMsg;
+
+  if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
+    // create tsma meta in dstVgId
+    if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) {
+      return -1;
+    }
+
+    // create stable to save tsma result in dstVgId
+    SVCreateStbReq pReq = {0};
+    pReq.name = pCfg->dstTbName;
+    pReq.suid = pCfg->dstTbUid;
+    pReq.schemaRow = pCfg->schemaRow;
+    pReq.schemaTag = pCfg->schemaTag;
+
+    if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) {
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
+/**
+ * @brief Insert/Update Time-range-wise SMA data.
+ *
+ * @param pSma
+ * @param msg
+ * @return int32_t
+ */
+int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
+  const SArray *pDataBlocks = (const SArray *)msg;
+  // TODO: destroy SSDataBlocks(msg)
+  if (!pDataBlocks) {
+    terrno = TSDB_CODE_TSMA_INVALID_PTR;
+    smaWarn("vgId:%d, insert tsma data failed since pDataBlocks is NULL", SMA_VID(pSma));
+    return terrno;
+  }
+
+  if (taosArrayGetSize(pDataBlocks) <= 0) {
+    terrno = TSDB_CODE_TSMA_INVALID_PARA;
+    smaWarn("vgId:%d, insert tsma data failed since pDataBlocks is empty", SMA_VID(pSma));
+    return TSDB_CODE_FAILED;
+  }
+
+  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE) != 0) {
+    terrno = TSDB_CODE_TSMA_INIT_FAILED;
+    return TSDB_CODE_FAILED;
+  }
+
+  SSmaEnv      *pEnv = SMA_TSMA_ENV(pSma);
+  SSmaStat     *pStat = NULL;
+  SSmaStatItem *pItem = NULL;
+
+  if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) {
+    terrno = TSDB_CODE_TSMA_INVALID_STAT;
+    return TSDB_CODE_FAILED;
+  }
+
+  tdRefSmaStat(pSma, pStat);
+  pItem = &pStat->tsmaStatItem;
+
+  ASSERT(pItem);
+
+  if (!pItem->pTSma) {
+    // cache smaMeta
+    STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid);
+    if (!pTSma) {
+      terrno = TSDB_CODE_TSMA_NO_INDEX_IN_META;
+      smaWarn("vgId:%d, tsma insert for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), indexUid, tstrerror(terrno));
+      return TSDB_CODE_FAILED;
+    }
+    pItem->pTSma = pTSma;
+  }
+
+  STSma *pTSma = pItem->pTSma;
+
+  ASSERT(pTSma->indexUid == indexUid);
+
+  SMetaReader mr = {0};
+
+  const char *dbName = "testDb";
+  if (metaGetTableEntryByName(&mr, dbName) != 0) {
+    smaDebug("vgId:%d, tsma no table testTb exists for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), indexUid, tstrerror(terrno));
+    SVCreateStbReq pReq = {0};
+    pReq.name = dbName;
+    pReq.suid = pTSma->dstTbUid;
+    pReq.schemaRow = pCfg->schemaRow;
+    pReq.schemaTag = pCfg->schemaTag;
+  }
+
+  SSubmitReq *pSubmitReq = NULL;
+  buildSubmitReqFromDataBlock(&pSubmitReq, (const SArray *)msg, NULL, pItem->pTSma->dstVgId,
+                              pItem->pTSma->dstTbUid);
+
+  tdUnRefSmaStat(pSma, pStat);
+
+  return TSDB_CODE_SUCCESS;
+}
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -110,6 +110,8 @@ int32_t tsdbBegin(STsdb *pTsdb) {
 }
 
 int32_t tsdbCommit(STsdb *pTsdb) {
+  if (!pTsdb) return 0;
+  
   int32_t    code = 0;
   SCommitH   commith = {0};
   SDFileSet *pSet = NULL;

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -20,6 +20,7 @@ extern const char *TSDB_LEVEL_DNAME[];
 typedef enum { TSDB_TXN_TEMP_FILE = 0, TSDB_TXN_CURR_FILE } TSDB_TXN_FILE_T;
 static const char *tsdbTxnFname[] = {"current.t", "current"};
 #define TSDB_MAX_FSETS(keep, days) ((keep) / (days) + 3)
+#define TSDB_MAX_INIT_FSETS        (365000)
 
 static int  tsdbComparFidFSet(const void *arg1, const void *arg2);
 static void tsdbResetFSStatus(SFSStatus *pStatus);
@@ -210,6 +211,10 @@ STsdbFS *tsdbNewFS(const STsdbKeepCfg *pCfg) {
     return NULL;
   }
 
+  if (maxFSet > TSDB_MAX_INIT_FSETS) {
+    maxFSet = TSDB_MAX_INIT_FSETS;
+  }
+
   pfs->cstatus = tsdbNewFSStatus(maxFSet);
   if (pfs->cstatus == NULL) {
     tsdbFreeFS(pfs);

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

@@ -289,7 +289,7 @@ void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
 }
 
 int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  int32_t ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+  int32_t ret = TAOS_SYNC_OTHER_ERROR;
 
   if (syncEnvIsStart()) {
     SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
@@ -370,13 +370,13 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 
     } else {
       vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
-      ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+      ret = TAOS_SYNC_OTHER_ERROR;
     }
 
     syncNodeRelease(pSyncNode);
   } else {
     vError("==vnodeProcessSyncReq== error syncEnv stop");
-    ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    ret = TAOS_SYNC_OTHER_ERROR;
   }
 
   return ret;

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

@@ -283,11 +283,14 @@ int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
 
 void vnodeSyncStart(SVnode *pVnode) {
   syncSetMsgCb(pVnode->sync, &pVnode->msgCb);
+  syncStart(pVnode->sync);
+  /*
   if (pVnode->config.standby) {
     syncStartStandBy(pVnode->sync);
   } else {
     syncStart(pVnode->sync);
   }
+  */
 }
 
 void vnodeSyncClose(SVnode *pVnode) { syncStop(pVnode->sync); }

source/libs/catalog/inc/catalogInt.h

@@ -57,6 +57,8 @@ enum {
   CTG_OP_DROP_TB_META,
   CTG_OP_UPDATE_USER,
   CTG_OP_UPDATE_VG_EPSET,
+  CTG_OP_UPDATE_TB_INDEX,
+  CTG_OP_DROP_TB_INDEX,
   CTG_OP_MAX
 };
 
@@ -128,25 +130,33 @@ typedef struct SCtgUserCtx {
   SUserAuthInfo user;
 } SCtgUserCtx;
 
-typedef struct SCtgTbMetaCache {
-  SRWLatch  stbLock;
-  SRWLatch  metaLock;        // RC between cache destroy and all other operations
-  SHashObj *metaCache;       //key:tbname, value:STableMeta
-  SHashObj *stbCache;        //key:suid, value:STableMeta*
-} SCtgTbMetaCache;
+typedef STableIndexRsp STableIndex;
 
-typedef struct SCtgDBCache {
+typedef struct SCtgTbCache {
+  SRWLatch     metaLock;
+  STableMeta  *pMeta;
+  SRWLatch     indexLock;
+  STableIndex *pIndex;
+} SCtgTbCache;
+
+typedef struct SCtgVgCache {
   SRWLatch         vgLock;
+  SDBVgInfo       *vgInfo;  
+} SCtgVgCache;
+
+typedef struct SCtgDBCache {
+  SRWLatch         dbLock;          // RC between destroy tbCache/stbCache and all reads
   uint64_t         dbId;
   int8_t           deleted;
-  SDBVgInfo       *vgInfo;  
-  SCtgTbMetaCache  tbCache;
+  SCtgVgCache      vgCache;
+  SHashObj        *tbCache;         // key:tbname, value:SCtgTbCache
+  SHashObj        *stbCache;        // key:suid, value:STableMeta*
 } SCtgDBCache;
 
 typedef struct SCtgRentSlot {
   SRWLatch lock;
   bool     needSort;
-  SArray  *meta;  // element is SDbVgVersion or SSTableMetaVersion
+  SArray  *meta;  // element is SDbVgVersion or SSTableVersion
 } SCtgRentSlot;
 
 typedef struct SCtgRentMgmt {
@@ -245,8 +255,10 @@ typedef struct SCtgCacheStat {
   uint64_t userNum;
   uint64_t vgHitNum;
   uint64_t vgMissNum;
-  uint64_t tblHitNum;
-  uint64_t tblMissNum;
+  uint64_t tbMetaHitNum;
+  uint64_t tbMetaMissNum;
+  uint64_t tbIndexHitNum;
+  uint64_t tbIndexMissNum;  
   uint64_t userHitNum;
   uint64_t userMissNum;
 } SCtgCacheStat;
@@ -268,10 +280,10 @@ typedef struct SCtgUpdateVgMsg {
   SDBVgInfo* dbInfo;
 } SCtgUpdateVgMsg;
 
-typedef struct SCtgUpdateTblMsg {
-  SCatalog* pCtg;
-  STableMetaOutput* output;
-} SCtgUpdateTblMsg;
+typedef struct SCtgUpdateTbMetaMsg {
+  SCatalog*         pCtg;
+  STableMetaOutput* pMeta;
+} SCtgUpdateTbMetaMsg;
 
 typedef struct SCtgDropDBMsg {
   SCatalog* pCtg;
@@ -305,6 +317,17 @@ typedef struct SCtgUpdateUserMsg {
   SGetUserAuthRsp userAuth;
 } SCtgUpdateUserMsg;
 
+typedef struct SCtgUpdateTbIndexMsg {
+  SCatalog*    pCtg;
+  STableIndex* pIndex;
+} SCtgUpdateTbIndexMsg;
+
+typedef struct SCtgDropTbIndexMsg {
+  SCatalog*    pCtg;
+  char         dbFName[TSDB_DB_FNAME_LEN];
+  char         tbName[TSDB_TABLE_NAME_LEN];
+} SCtgDropTbIndexMsg;
+
 typedef struct SCtgUpdateEpsetMsg {
   SCatalog* pCtg;
   char  dbFName[TSDB_DB_FNAME_LEN];
@@ -465,12 +488,11 @@ int32_t ctgOpUpdateUser(SCtgCacheOperation *action);
 int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation);
 int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
 void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache);
-void ctgReleaseVgInfo(SCtgDBCache *dbCache);
-int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
+void ctgRUnlockVgInfo(SCtgDBCache *dbCache);
 int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist);
 int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
-int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
-int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
+int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
+int32_t ctgChkAuthFromCache(SCatalog* pCtg, char* user, char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
 int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
 int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncReq);
 int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
@@ -479,12 +501,18 @@ int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId
 int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
 int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
 int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet);
+int32_t ctgUpdateTbIndexEnqueue(SCatalog* pCtg, STableIndex **pIndex, bool syncOp);
 int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type);
 int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size);
 int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size);
 int32_t ctgUpdateTbMetaToCache(SCatalog* pCtg, STableMetaOutput* pOut, bool syncReq);
 int32_t ctgStartUpdateThread();
 int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask);
+void ctgReleaseVgInfoToCache(SCatalog* pCtg, SCtgDBCache *dbCache);
+int32_t ctgReadTbIndexFromCache(SCatalog* pCtg, SName* pTableName, SArray** pRes);
+int32_t ctgDropTbIndexEnqueue(SCatalog* pCtg, SName* pName, bool syncOp);
+int32_t ctgOpDropTbIndex(SCtgCacheOperation *operation);
+int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation);
 
 
 
@@ -493,7 +521,7 @@ int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SBuildU
 int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SArray *out, SCtgTask* pTask);
 int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
 int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *indexName, SIndexInfo *out, SCtgTask* pTask);
-int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName* name, SArray** out, SCtgTask* pTask);
+int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *name, STableIndex* out, SCtgTask* pTask);
 int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
 int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
 int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
@@ -521,6 +549,8 @@ void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput);
 int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target);
 char *ctgTaskTypeStr(CTG_TASK_TYPE type);
 int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask);
+int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes);
+void ctgFreeSTableIndex(void *info);
 
 
 extern SCatalogMgmt gCtgMgmt;

source/libs/catalog/src/catalog.c

@@ -96,8 +96,7 @@ int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
   if (NULL != dbCache) {
     input.dbId = dbCache->dbId;
 
-    ctgReleaseVgInfo(dbCache);
-    ctgReleaseDBCache(pCtg, dbCache);
+    ctgReleaseVgInfoToCache(pCtg, dbCache);
   }
 
   input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
@@ -349,8 +348,8 @@ int32_t ctgChkAuth(SCatalog* pCtg, SRequestConnInfo *pConn, const char* user, co
   int32_t code = 0;
 
   *pass = false;
-
-  CTG_ERR_RET(ctgChkAuthFromCache(pCtg, user, dbFName, type, &inCache, pass));
+  
+  CTG_ERR_RET(ctgChkAuthFromCache(pCtg, (char*)user, (char*)dbFName, type, &inCache, pass));
 
   if (inCache) {
     return TSDB_CODE_SUCCESS;
@@ -382,6 +381,45 @@ _return:
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgGetTbIndex(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pRes) {
+  CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes));
+  if (*pRes) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  STableIndex *pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
+  if (NULL == pIndex) {
+    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+  }
+  
+  int32_t code = ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pIndex, NULL);
+  if (TSDB_CODE_MND_DB_INDEX_NOT_EXIST == code) {
+    code = 0;
+    goto _return;
+  }
+  CTG_ERR_JRET(code);
+  
+  SArray* pInfo = NULL;
+  CTG_ERR_JRET(ctgCloneTableIndex(pIndex->pIndex, &pInfo));
+
+  *pRes = pInfo;
+
+  CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, &pIndex, false));
+
+  return TSDB_CODE_SUCCESS;
+
+_return:
+
+  tFreeSTableIndexRsp(pIndex);
+  taosMemoryFree(pIndex);
+
+  taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
+  *pRes = NULL;
+
+  CTG_RET(code);
+}
+
+
 int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pVgList) {
   STableMeta *tbMeta = NULL;
   int32_t code = 0;
@@ -404,7 +442,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTabl
   CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo));
 
   if (dbCache) {
-    vgHash = dbCache->vgInfo->vgHash;
+    vgHash = dbCache->vgCache.vgInfo->vgHash;
   } else {
     vgHash = vgInfo->vgHash;
   }
@@ -442,7 +480,7 @@ int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTabl
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
+    ctgRUnlockVgInfo(dbCache);
     ctgReleaseDBCache(pCtg, dbCache);
   }
 
@@ -631,12 +669,11 @@ int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* vers
     CTG_API_LEAVE(TSDB_CODE_SUCCESS);
   }
 
-  *version = dbCache->vgInfo->vgVersion;
+  *version = dbCache->vgCache.vgInfo->vgVersion;
   *dbId = dbCache->dbId;
-  *tableNum = dbCache->vgInfo->numOfTable;
+  *tableNum = dbCache->vgCache.vgInfo->numOfTable;
 
-  ctgReleaseVgInfo(dbCache);
-  ctgReleaseDBCache(pCtg, dbCache);
+  ctgReleaseVgInfoToCache(pCtg, dbCache);
 
   ctgDebug("Got db vgVersion from cache, dbFName:%s, vgVersion:%d", dbFName, *version);
 
@@ -661,7 +698,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
   SDBVgInfo *vgInfo = NULL;
   CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, dbFName, &dbCache, &vgInfo));
   if (dbCache) {
-    vgHash = dbCache->vgInfo->vgHash;
+    vgHash = dbCache->vgCache.vgInfo->vgHash;
   } else {
     vgHash = vgInfo->vgHash;
   }
@@ -674,7 +711,7 @@ int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char*
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
+    ctgRUnlockVgInfo(dbCache);
     ctgReleaseDBCache(pCtg, dbCache);
   }
 
@@ -741,6 +778,30 @@ _return:
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogUpdateTableIndex(SCatalog* pCtg, STableIndexRsp *pRsp) {
+  CTG_API_ENTER();
+
+  int32_t code = 0;
+  
+  if (NULL == pCtg || NULL == pRsp) {
+    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
+  }
+
+  STableIndex* pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
+  if (NULL == pIndex) {
+    CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
+  }
+
+  memcpy(pIndex, pRsp, sizeof(STableIndex));
+
+  CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pCtg, &pIndex, false));
+
+_return:
+  
+  CTG_API_LEAVE(code);
+}
+
+
 int32_t catalogRemoveTableMeta(SCatalog* pCtg, SName* pTableName) {
   CTG_API_ENTER();
 
@@ -932,12 +993,12 @@ int32_t catalogGetTableHashVgroup(SCatalog *pCtg, SRequestConnInfo *pConn, const
   SDBVgInfo *vgInfo = NULL;
   CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, db, &dbCache, &vgInfo));
 
-  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, vgInfo ? vgInfo : dbCache->vgInfo, pTableName, pVgroup));
+  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, vgInfo ? vgInfo : dbCache->vgCache.vgInfo, pTableName, pVgroup));
 
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
+    ctgRUnlockVgInfo(dbCache);
     ctgReleaseDBCache(pCtg, dbCache);
   }
 
@@ -1060,14 +1121,14 @@ _return:
   CTG_API_LEAVE(TSDB_CODE_SUCCESS);
 }
 
-int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion** stables, uint32_t* num) {
+int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableVersion **stables, uint32_t *num) {
   CTG_API_ENTER();
 
   if (NULL == pCtg || NULL == stables || NULL == num) {
     CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
-  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void**)stables, num, sizeof(SSTableMetaVersion)));
+  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableVersion)));
 }
 
 int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion** dbs, uint32_t* num) {
@@ -1138,7 +1199,12 @@ int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo *pConn, const SNam
     CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
-  CTG_API_LEAVE(ctgGetTbIndexFromMnode(pCtg, pConn, (SName*)pTableName, pRes, NULL));
+  int32_t code = 0;
+  CTG_ERR_JRET(ctgGetTbIndex(pCtg, pConn, (SName*)pTableName, pRes));
+  
+_return:
+
+  CTG_API_LEAVE(code);
 }
 
 int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* funcName, SFuncInfo* pInfo) {

source/libs/catalog/src/ctgAsync.c

@@ -344,6 +344,11 @@ int32_t ctgHandleForceUpdate(SCatalog* pCtg, SCtgJob *pJob, const SCatalogReq* p
     }
   }
 
+  for (int32_t i = 0; i < pJob->tbIndexNum; ++i) {
+    SName* name = taosArrayGet(pReq->pTableIndex, i);
+    ctgDropTbIndexEnqueue(pCtg, name, true);
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -680,15 +685,14 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
           CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
           if (NULL != dbCache) {
             SVgroupInfo vgInfo = {0};
-            CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo));
+            CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, ctx->pName, &vgInfo));
           
             ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(ctx->pName), ctx->flag);
 
             ctx->vgId = vgInfo.vgId;          
             CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, ctx->pName, &vgInfo, NULL, pTask));
 
-            ctgReleaseVgInfo(dbCache);
-            ctgReleaseDBCache(pCtg, dbCache);
+            ctgReleaseVgInfoToCache(pCtg, dbCache);
           } else {
             SBuildUseDBInput input = {0};
           
@@ -786,7 +790,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
+    ctgRUnlockVgInfo(dbCache);
     ctgReleaseDBCache(pCtg, dbCache);
   }
 
@@ -866,9 +870,15 @@ _return:
 
 int32_t ctgHandleGetTbIndexRsp(SCtgTask* pTask, int32_t reqType, const SDataBuf *pMsg, int32_t rspCode) {
   int32_t code = 0;
-  CTG_ERR_JRET(ctgProcessRspMsg(pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
+  CTG_ERR_JRET(ctgProcessRspMsg(&pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target));
 
-  TSWAP(pTask->res, pTask->msgCtx.out);
+  STableIndex* pOut = (STableIndex*)pTask->msgCtx.out;
+  SArray* pInfo = NULL;
+  CTG_ERR_JRET(ctgCloneTableIndex(pOut->pIndex, &pInfo));
+  pTask->res = pInfo;
+
+  SCtgTbIndexCtx* ctx = pTask->taskCtx;  
+  CTG_ERR_JRET(ctgUpdateTbIndexEnqueue(pTask->pJob->pCtg, (STableIndex**)&pTask->msgCtx.out, false));
   
 _return:
   if (TSDB_CODE_MND_DB_INDEX_NOT_EXIST == code) {
@@ -1008,7 +1018,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
   CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
   if (dbCache) {
     SVgroupInfo vgInfo = {0};
-    CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, ctx->pName, &vgInfo));
+    CTG_ERR_RET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, ctx->pName, &vgInfo));
 
     ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(ctx->pName), ctx->flag);
 
@@ -1026,8 +1036,7 @@ int32_t ctgAsyncRefreshTbMeta(SCtgTask *pTask) {
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
-    ctgReleaseDBCache(pCtg, dbCache);
+    ctgReleaseVgInfoToCache(pCtg, dbCache);
   }
 
   CTG_RET(code);
@@ -1057,7 +1066,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) {
   
   CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache));
   if (NULL != dbCache) {
-    CTG_ERR_JRET(ctgGenerateVgList(pCtg, dbCache->vgInfo->vgHash, (SArray**)&pTask->res));
+    CTG_ERR_JRET(ctgGenerateVgList(pCtg, dbCache->vgCache.vgInfo->vgHash, (SArray**)&pTask->res));
     
     CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0));
   } else {
@@ -1072,8 +1081,7 @@ int32_t ctgLaunchGetDbVgTask(SCtgTask *pTask) {
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
-    ctgReleaseDBCache(pCtg, dbCache);
+    ctgReleaseVgInfoToCache(pCtg, dbCache);
   }
 
   CTG_RET(code);
@@ -1092,7 +1100,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) {
     if (NULL == pTask->res) {
       CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
     }
-    CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgInfo, pCtx->pName, (SVgroupInfo*)pTask->res));
+    CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pCtx->pName, (SVgroupInfo*)pTask->res));
     
     CTG_ERR_JRET(ctgHandleTaskEnd(pTask, 0));
   } else {
@@ -1107,8 +1115,7 @@ int32_t ctgLaunchGetTbHashTask(SCtgTask *pTask) {
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
-    ctgReleaseDBCache(pCtg, dbCache);
+    ctgReleaseVgInfoToCache(pCtg, dbCache);
   }
 
   CTG_RET(code);
@@ -1119,6 +1126,15 @@ int32_t ctgLaunchGetTbIndexTask(SCtgTask *pTask) {
   SCatalog* pCtg = pTask->pJob->pCtg; 
   SRequestConnInfo* pConn = &pTask->pJob->conn;
   SCtgTbIndexCtx* pCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
+  SArray* pRes = NULL;
+
+  CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pCtx->pName, &pRes));
+  if (pRes) {
+    pTask->res = pRes;
+    
+    CTG_ERR_RET(ctgHandleTaskEnd(pTask, 0));
+    return TSDB_CODE_SUCCESS;
+  }
   
   CTG_ERR_RET(ctgGetTbIndexFromMnode(pCtg, pConn, pCtx->pName, NULL, pTask));
   return TSDB_CODE_SUCCESS;
@@ -1157,9 +1173,9 @@ int32_t ctgLaunchGetDbInfoTask(SCtgTask *pTask) {
   SDbInfo* pInfo = (SDbInfo*)pTask->res;
   CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, pCtx->dbFName, &dbCache));
   if (NULL != dbCache) {
-    pInfo->vgVer = dbCache->vgInfo->vgVersion;
+    pInfo->vgVer = dbCache->vgCache.vgInfo->vgVersion;
     pInfo->dbId = dbCache->dbId;
-    pInfo->tbNum = dbCache->vgInfo->numOfTable;
+    pInfo->tbNum = dbCache->vgCache.vgInfo->numOfTable;
   } else {
     pInfo->vgVer = CTG_DEFAULT_INVALID_VERSION;
   }
@@ -1169,8 +1185,7 @@ int32_t ctgLaunchGetDbInfoTask(SCtgTask *pTask) {
 _return:
 
   if (dbCache) {
-    ctgReleaseVgInfo(dbCache);
-    ctgReleaseDBCache(pCtg, dbCache);
+    ctgReleaseVgInfoToCache(pCtg, dbCache);
   }
 
   CTG_RET(code);

source/libs/catalog/src/ctgDbg.c

@@ -19,7 +19,7 @@
 #include "catalogInt.h"
 
 extern SCatalogMgmt gCtgMgmt;
-SCtgDebug gCTGDebug = {.apiEnable = true};
+SCtgDebug gCTGDebug = {0};
 
 void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) {
   ASSERT(*(int32_t*)param == 1);
@@ -266,11 +266,11 @@ int32_t ctgdGetStatNum(char *option, void *res) {
 }
 
 int32_t ctgdGetTbMetaNum(SCtgDBCache *dbCache) {
-  return dbCache->tbCache.metaCache ? (int32_t)taosHashGetSize(dbCache->tbCache.metaCache) : 0;
+  return dbCache->tbCache ? (int32_t)taosHashGetSize(dbCache->tbCache) : 0;
 }
 
 int32_t ctgdGetStbNum(SCtgDBCache *dbCache) {
-  return dbCache->tbCache.stbCache ? (int32_t)taosHashGetSize(dbCache->tbCache.stbCache) : 0;
+  return dbCache->stbCache ? (int32_t)taosHashGetSize(dbCache->stbCache) : 0;
 }
 
 int32_t ctgdGetRentNum(SCtgRentMgmt *rent) {
@@ -363,17 +363,17 @@ void ctgdShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
 
     dbFName = taosHashGetKey(pIter, &len);
 
-    int32_t metaNum = dbCache->tbCache.metaCache ? taosHashGetSize(dbCache->tbCache.metaCache) : 0;
-    int32_t stbNum = dbCache->tbCache.stbCache ? taosHashGetSize(dbCache->tbCache.stbCache) : 0;
+    int32_t metaNum = dbCache->tbCache ? taosHashGetSize(dbCache->tbCache) : 0;
+    int32_t stbNum = dbCache->stbCache ? taosHashGetSize(dbCache->stbCache) : 0;
     int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION;
     int32_t hashMethod = -1;
     int32_t vgNum = 0;
 
-    if (dbCache->vgInfo) {
-      vgVersion = dbCache->vgInfo->vgVersion;
-      hashMethod = dbCache->vgInfo->hashMethod;
-      if (dbCache->vgInfo->vgHash) {
-        vgNum = taosHashGetSize(dbCache->vgInfo->vgHash);
+    if (dbCache->vgCache.vgInfo) {
+      vgVersion = dbCache->vgCache.vgInfo->vgVersion;
+      hashMethod = dbCache->vgCache.vgInfo->hashMethod;
+      if (dbCache->vgCache.vgInfo->vgHash) {
+        vgNum = taosHashGetSize(dbCache->vgCache.vgInfo->vgHash);
       }
     }
     

source/libs/catalog/src/ctgRemote.c

@@ -431,7 +431,7 @@ int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *name, SArray** out, SCtgTask* pTask) {
+int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *name, STableIndex* out, SCtgTask* pTask) {
   char *msg = NULL;
   int32_t msgLen = 0;
   int32_t reqType = TDMT_MND_GET_TABLE_INDEX;
@@ -448,10 +448,11 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, SName *n
   }
 
   if (pTask) {
-    void* pOut = taosMemoryCalloc(1, POINTER_BYTES);
+    void* pOut = taosMemoryCalloc(1, sizeof(STableIndex));
     if (NULL == pOut) {
       CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
     }
+    
     CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, pOut, (char*)tbFName));
     
     CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask, reqType, msg, msgLen));

source/libs/catalog/src/ctgUtil.c

@@ -44,6 +44,16 @@ char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
   }
 }
 
+void ctgFreeSTableIndex(void *info) {
+  if (NULL == info) {
+    return;
+  }
+
+  STableIndex *pInfo = (STableIndex *)info;
+
+  taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo);
+}
+
 void ctgFreeSMetaData(SMetaData* pData) {
   taosArrayDestroy(pData->pTableMeta);
   pData->pTableMeta = NULL;
@@ -110,25 +120,39 @@ void ctgFreeMetaRent(SCtgRentMgmt *mgmt) {
   taosMemoryFreeClear(mgmt->slots);
 }
 
+void ctgFreeStbMetaCache(SCtgDBCache *dbCache) {
+  if (NULL == dbCache->stbCache) {
+    return;
+  }
 
-void ctgFreeTbMetaCache(SCtgTbMetaCache *cache) {
-  CTG_LOCK(CTG_WRITE, &cache->stbLock);
-  if (cache->stbCache) {
-    int32_t stblNum = taosHashGetSize(cache->stbCache);  
-    taosHashCleanup(cache->stbCache);
-    cache->stbCache = NULL;
-    CTG_CACHE_STAT_DEC(stblNum, stblNum);
+  int32_t stblNum = taosHashGetSize(dbCache->stbCache);  
+  taosHashCleanup(dbCache->stbCache);
+  dbCache->stbCache = NULL;
+  CTG_CACHE_STAT_DEC(stblNum, stblNum);
+}
+
+void ctgFreeTbCacheImpl(SCtgTbCache *pCache) {
+  taosMemoryFreeClear(pCache->pMeta);
+  if (pCache->pIndex) {
+    taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo);
+    taosMemoryFreeClear(pCache->pIndex);
+  }
+}
+
+void ctgFreeTbCache(SCtgDBCache *dbCache) {
+  if (NULL == dbCache->tbCache) {
+    return;
   }
-  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
 
-  CTG_LOCK(CTG_WRITE, &cache->metaLock);
-  if (cache->metaCache) {
-    int32_t tblNum = taosHashGetSize(cache->metaCache);
-    taosHashCleanup(cache->metaCache);
-    cache->metaCache = NULL;
-    CTG_CACHE_STAT_DEC(tblNum, tblNum);
+  int32_t tblNum = taosHashGetSize(dbCache->tbCache);
+  SCtgTbCache *pCache = taosHashIterate(dbCache->tbCache, NULL);
+  while (NULL != pCache) {
+    ctgFreeTbCacheImpl(pCache);
+    pCache = taosHashIterate(dbCache->tbCache, pCache);
   }
-  CTG_UNLOCK(CTG_WRITE, &cache->metaLock);
+  taosHashCleanup(dbCache->tbCache);
+  dbCache->tbCache = NULL;
+  CTG_CACHE_STAT_DEC(tblNum, tblNum);
 }
 
 void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
@@ -144,16 +168,18 @@ void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
   taosMemoryFreeClear(vgInfo);
 }
 
+void ctgFreeVgInfoCache(SCtgDBCache *dbCache) {
+  ctgFreeVgInfo(dbCache->vgCache.vgInfo);
+}
+
 void ctgFreeDbCache(SCtgDBCache *dbCache) {
   if (NULL == dbCache) {
     return;
   }
 
-  CTG_LOCK(CTG_WRITE, &dbCache->vgLock);
-  ctgFreeVgInfo (dbCache->vgInfo);
-  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
-
-  ctgFreeTbMetaCache(&dbCache->tbCache);
+  ctgFreeVgInfoCache(dbCache);
+  ctgFreeStbMetaCache(dbCache);
+  ctgFreeTbCache(dbCache);
 }
 
 
@@ -167,16 +193,13 @@ void ctgFreeHandle(SCatalog* pCtg) {
     void *pIter = taosHashIterate(pCtg->dbCache, NULL);
     while (pIter) {
       SCtgDBCache *dbCache = pIter;
-
       atomic_store_8(&dbCache->deleted, 1);
-
       ctgFreeDbCache(dbCache);
             
       pIter = taosHashIterate(pCtg->dbCache, pIter);
     }  
 
     taosHashCleanup(pCtg->dbCache);
-    
     CTG_CACHE_STAT_DEC(dbNum, dbNum);
   }
 
@@ -186,14 +209,12 @@ void ctgFreeHandle(SCatalog* pCtg) {
     void *pIter = taosHashIterate(pCtg->userCache, NULL);
     while (pIter) {
       SCtgUserAuth *userCache = pIter;
-
       ctgFreeSCtgUserAuth(userCache);
 
       pIter = taosHashIterate(pCtg->userCache, pIter);
     }  
 
     taosHashCleanup(pCtg->userCache);
-
     CTG_CACHE_STAT_DEC(userNum, userNum);
   }
 
@@ -252,9 +273,9 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
       break;
     }
     case TDMT_MND_GET_TABLE_INDEX: {
-      SArray** pOut = (SArray**)pCtx->out;
+      STableIndex* pOut = (STableIndex*)pCtx->out;
       if (pOut) {
-        taosArrayDestroyEx(*pOut, tFreeSTableIndexInfo);
+        taosArrayDestroyEx(pOut->pIndex, tFreeSTableIndexInfo);
         taosMemoryFreeClear(pCtx->out);
       }
       break;
@@ -535,9 +556,9 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName
 }
 
 int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
-  if (*(uint64_t *)key1 < ((SSTableMetaVersion*)key2)->suid) {
+  if (*(uint64_t *)key1 < ((SSTableVersion*)key2)->suid) {
     return -1;
-  } else if (*(uint64_t *)key1 > ((SSTableMetaVersion*)key2)->suid) {
+  } else if (*(uint64_t *)key1 > ((SSTableVersion*)key2)->suid) {
     return 1;
   } else {
     return 0;
@@ -555,9 +576,9 @@ int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2) {
 }
 
 int32_t ctgStbVersionSortCompare(const void* key1, const void* key2) {
-  if (((SSTableMetaVersion*)key1)->suid < ((SSTableMetaVersion*)key2)->suid) {
+  if (((SSTableVersion*)key1)->suid < ((SSTableVersion*)key2)->suid) {
     return -1;
-  } else if (((SSTableMetaVersion*)key1)->suid > ((SSTableMetaVersion*)key2)->suid) {
+  } else if (((SSTableVersion*)key1)->suid > ((SSTableVersion*)key2)->suid) {
     return 1;
   } else {
     return 0;
@@ -640,6 +661,27 @@ int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput)
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
+  if (NULL == pIndex) {
+    *pRes = NULL;
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int32_t num = taosArrayGetSize(pIndex);
+  *pRes = taosArrayInit(num, sizeof(STableIndexInfo));
+  if (NULL == *pRes) {
+    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+  }
+
+  for (int32_t i = 0; i < num; ++i) {
+    STableIndexInfo *pInfo = taosArrayGet(pIndex, i);
+    taosArrayPush(*pRes, pInfo);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask) {
   if (msgType == TDMT_VND_TABLE_META) {
     SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;

source/libs/catalog/test/catalogTests.cpp

@@ -895,9 +895,9 @@ void *ctgTestSetCtableMetaThread(void *param) {
     output = (STableMetaOutput *)taosMemoryMalloc(sizeof(STableMetaOutput));
     ctgTestBuildCTableMetaOutput(output);
 
-    SCtgUpdateTblMsg *msg = (SCtgUpdateTblMsg *)taosMemoryMalloc(sizeof(SCtgUpdateTblMsg));
+    SCtgUpdateTbMetaMsg *msg = (SCtgUpdateTbMetaMsg *)taosMemoryMalloc(sizeof(SCtgUpdateTbMetaMsg));
     msg->pCtg = pCtg;
-    msg->output = output;
+    msg->pMeta = output;
     operation.data = msg;
 
     code = ctgOpUpdateTbMeta(&operation);
@@ -989,7 +989,7 @@ TEST(tableMeta, normalTable) {
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
   SDbVgVersion       *dbs = NULL;
-  SSTableMetaVersion *stb = NULL;
+  SSTableVersion *stb = NULL;
   uint32_t            dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
   int32_t             i = 0;
   while (i < 5) {
@@ -1098,7 +1098,7 @@ TEST(tableMeta, childTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
   SDbVgVersion       *dbs = NULL;
-  SSTableMetaVersion *stb = NULL;
+  SSTableVersion *stb = NULL;
   uint32_t            dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
   int32_t             i = 0;
   while (i < 5) {
@@ -1220,7 +1220,7 @@ TEST(tableMeta, superTableCase) {
   ASSERT_EQ(tableMeta->tableInfo.rowSize, 12);
 
   SDbVgVersion       *dbs = NULL;
-  SSTableMetaVersion *stb = NULL;
+  SSTableVersion *stb = NULL;
   uint32_t            dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0;
   int32_t             i = 0;
   while (i < 5) {
@@ -2299,7 +2299,7 @@ TEST(rentTest, allRent) {
   SArray          *vgList = NULL;
   ctgTestStop = false;
   SDbVgVersion       *dbs = NULL;
-  SSTableMetaVersion *stable = NULL;
+  SSTableVersion *stable = NULL;
   uint32_t            num = 0;
 
   ctgTestInitLogFile();

source/libs/index/src/indexFilter.c

@@ -97,7 +97,14 @@ static int32_t sifGetOperParamNum(EOperatorType ty) {
   }
   return 2;
 }
-static int32_t sifValidateColumn(SColumnNode *cn) {
+static int32_t sifValidOp(EOperatorType ty) {
+  if ((ty >= OP_TYPE_ADD && ty <= OP_TYPE_BIT_OR) || (ty == OP_TYPE_IN || ty == OP_TYPE_NOT_IN) ||
+      (ty == OP_TYPE_LIKE || ty == OP_TYPE_NOT_LIKE || ty == OP_TYPE_MATCH || ty == OP_TYPE_NMATCH)) {
+    return -1;
+  }
+  return 0;
+}
+static int32_t sifValidColumn(SColumnNode *cn) {
   // add more check
   if (cn == NULL) {
     return TSDB_CODE_QRY_INVALID_INPUT;
@@ -176,6 +183,7 @@ static int32_t sifInitJsonParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
   memcpy(param->colName, r->literal, strlen(r->literal));
   // sprintf(param->colName, "%s_%s", l->colName, r->literal);
   param->colValType = r->typeData;
+  param->status = SFLT_COARSE_INDEX;
   return 0;
   // memcpy(param->colName, l->colName, sizeof(l->colName));
 }
@@ -197,7 +205,7 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
     case QUERY_NODE_COLUMN: {
       SColumnNode *cn = (SColumnNode *)node;
       /*only support tag column*/
-      SIF_ERR_RET(sifValidateColumn(cn));
+      SIF_ERR_RET(sifValidColumn(cn));
 
       param->colId = cn->colId;
       param->colValType = cn->node.resType.type;
@@ -247,11 +255,13 @@ static int32_t sifInitOperParams(SIFParam **params, SOperatorNode *node, SIFCtx
     return code;
   }
   SIFParam *paramList = taosMemoryCalloc(nParam, sizeof(SIFParam));
+
   if (NULL == paramList) {
     SIF_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
   }
 
-  if (nodeType(node->pLeft) == QUERY_NODE_OPERATOR) {
+  if (nodeType(node->pLeft) == QUERY_NODE_OPERATOR &&
+      (((SOperatorNode *)(node->pLeft))->opType == OP_TYPE_JSON_GET_VALUE)) {
     SNode *interNode = (node->pLeft);
     SIF_ERR_JRET(sifInitJsonParam(interNode, &paramList[0], ctx));
     if (nParam > 1) {
@@ -505,6 +515,11 @@ static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *sta
 
 static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
   int32_t code = 0;
+  if (sifValidOp(node->opType) < 0) {
+    output->status = SFLT_NOT_INDEX;
+    return code;
+  }
+
   int32_t nParam = sifGetOperParamNum(node->opType);
   if (nParam <= 1) {
     output->status = SFLT_NOT_INDEX;
@@ -513,9 +528,15 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
   if (node->opType == OP_TYPE_JSON_GET_VALUE) {
     return code;
   }
-  SIFParam *params = NULL;
 
+  SIFParam *params = NULL;
   SIF_ERR_RET(sifInitOperParams(&params, node, ctx));
+
+  if (params[0].status == SFLT_NOT_INDEX || (nParam > 1 && params[1].status == SFLT_NOT_INDEX)) {
+    output->status = SFLT_NOT_INDEX;
+    return code;
+  }
+
   // ugly code, refactor later
   output->arg = ctx->arg;
   sif_func_t operFn = sifNullFunc;

source/libs/qcom/src/querymsg.c

@@ -484,13 +484,11 @@ int32_t queryProcessGetTbIndexRsp(void *output, char *msg, int32_t msgSize) {
     return TSDB_CODE_TSC_INVALID_INPUT;
   }
 
-  STableIndexRsp out = {0};
-  if (tDeserializeSTableIndexRsp(msg, msgSize, &out) != 0) {
+  STableIndexRsp *out = (STableIndexRsp*)output;
+  if (tDeserializeSTableIndexRsp(msg, msgSize, out) != 0) {
     qError("tDeserializeSTableIndexRsp failed, msgSize:%d", msgSize);
     return TSDB_CODE_INVALID_MSG;
   }
-
-  *(void **)output = out.pIndex;
   
   return TSDB_CODE_SUCCESS;
 }

source/libs/qworker/src/qworker.c

@@ -837,7 +837,6 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
   SQWorker *mgmt = qwAcquire(refId);
   if (NULL == mgmt) {
     QW_DLOG("qwAcquire %" PRIx64 "failed", refId);
-    taosMemoryFree(param);
     return;
   }
 

source/libs/scalar/src/filter.c

@@ -1476,6 +1476,11 @@ void filterDumpInfoToString(SFilterInfo *info, const char *msg, int32_t options)
       for (uint32_t i = 0; i < info->fields[FLD_TYPE_VALUE].num; ++i) {
         SFilterField *field = &info->fields[FLD_TYPE_VALUE].fields[i];
         if (field->desc) {
+          if (QUERY_NODE_VALUE != nodeType(field->desc)) {
+            qDebug("VAL%d => [type:not value node][val:NIL]", i); //TODO
+            continue;
+          }
+
           SValueNode *var = (SValueNode *)field->desc;
           SDataType *dType = &var->node.resType;
           if (dType->type == TSDB_DATA_TYPE_VALUE_ARRAY) {

source/libs/sync/inc/syncInt.h

@@ -168,6 +168,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo);
 void       syncNodeStart(SSyncNode* pSyncNode);
 void       syncNodeStartStandBy(SSyncNode* pSyncNode);
 void       syncNodeClose(SSyncNode* pSyncNode);
+int32_t    syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak);
 
 // option
 bool syncNodeSnapshotEnable(SSyncNode* pSyncNode);
@@ -232,6 +233,9 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
 bool                 syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId);
 SSyncSnapshotSender* syncNodeGetSnapshotSender(SSyncNode* ths, SRaftId* pDestId);
 
+void syncStartNormal(int64_t rid);
+void syncStartStandBy(int64_t rid);
+
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj);
 void syncNodePrint2(char* s, SSyncNode* pObj);

source/libs/sync/src/syncAppendEntries.c

@@ -995,7 +995,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
             ths->commitIndex = snapshot.lastApplyIndex;
 
             sDebug("vgId:%d sync event commit by snapshot from index:%ld to index:%ld, %s", ths->vgId, commitBegin,
-                  commitEnd, syncUtilState2String(ths->state));
+                   commitEnd, syncUtilState2String(ths->state));
           }
 
           SyncIndex beginIndex = ths->commitIndex + 1;

source/libs/sync/src/syncCommit.c

@@ -57,7 +57,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
     pSyncNode->commitIndex = snapshot.lastApplyIndex;
 
     sDebug("vgId:%d sync event commit by snapshot from index:%ld to index:%ld, %s", pSyncNode->vgId,
-          pSyncNode->commitIndex, snapshot.lastApplyIndex, syncUtilState2String(pSyncNode->state));
+           pSyncNode->commitIndex, snapshot.lastApplyIndex, syncUtilState2String(pSyncNode->state));
   }
 
   // update commit index

source/libs/sync/src/syncMain.c

@@ -149,12 +149,12 @@ void syncStop(int64_t rid) {
 int32_t syncSetStandby(int64_t rid) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return -1;
+    return TAOS_SYNC_OTHER_ERROR;
   }
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     taosReleaseRef(tsNodeRefId, pSyncNode->rid);
-    return -1;
+    return TAOS_SYNC_OTHER_ERROR;
   }
 
   // state change
@@ -174,14 +174,88 @@ int32_t syncSetStandby(int64_t rid) {
   return 0;
 }
 
-int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
+int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return TAOS_SYNC_OTHER_ERROR;
+  }
+  ASSERT(rid == pSyncNode->rid);
+
   int32_t ret = 0;
-  char*   newconfig = syncCfg2Str((SSyncCfg*)pSyncCfg);
+  bool    IamInNew = false;
+  for (int i = 0; i < pNewCfg->replicaNum; ++i) {
+    if (strcmp((pNewCfg->nodeInfo)[i].nodeFqdn, pSyncNode->myNodeInfo.nodeFqdn) == 0 &&
+        (pNewCfg->nodeInfo)[i].nodePort == pSyncNode->myNodeInfo.nodePort) {
+      IamInNew = true;
+    }
+
+    /*
+        SRaftId newId;
+        newId.addr = syncUtilAddr2U64((pNewCfg->nodeInfo)[i].nodeFqdn, (pNewCfg->nodeInfo)[i].nodePort);
+        newId.vgId = pSyncNode->vgId;
+        if (syncUtilSameId(&(pSyncNode->myRaftId), &newId)) {
+          IamInNew = true;
+        }
+    */
+  }
+
+  if (!IamInNew) {
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+    return TAOS_SYNC_NOT_IN_NEW_CONFIG;
+  }
+
+  char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
+  pRpcMsg->msgType = TDMT_SYNC_CONFIG_CHANGE;
+  pRpcMsg->info.noResp = 1;
+  pRpcMsg->contLen = strlen(newconfig) + 1;
+  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
+  snprintf(pRpcMsg->pCont, pRpcMsg->contLen, "%s", newconfig);
+  taosMemoryFree(newconfig);
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+  return ret;
+}
+
+int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return TAOS_SYNC_OTHER_ERROR;
+  }
+  ASSERT(rid == pSyncNode->rid);
+
+  bool IamInNew = false;
+  for (int i = 0; i < pNewCfg->replicaNum; ++i) {
+    if (strcmp((pNewCfg->nodeInfo)[i].nodeFqdn, pSyncNode->myNodeInfo.nodeFqdn) == 0 &&
+        (pNewCfg->nodeInfo)[i].nodePort == pSyncNode->myNodeInfo.nodePort) {
+      IamInNew = true;
+    }
+
+    /*
+        // some problem in inet_addr
+
+        SRaftId newId = EMPTY_RAFT_ID;
+        newId.addr = syncUtilAddr2U64((pNewCfg->nodeInfo)[i].nodeFqdn, (pNewCfg->nodeInfo)[i].nodePort);
+        newId.vgId = pSyncNode->vgId;
+
+        if (syncUtilSameId(&(pSyncNode->myRaftId), &newId)) {
+          IamInNew = true;
+        }
+      */
+  }
+
+  if (!IamInNew) {
+    sError("sync reconfig error, not in new config");
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+    return TAOS_SYNC_NOT_IN_NEW_CONFIG;
+  }
 
+  char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
   if (gRaftDetailLog) {
     sInfo("==syncReconfig== newconfig:%s", newconfig);
   }
 
+  int32_t ret = 0;
+
   SRpcMsg rpcMsg = {0};
   rpcMsg.msgType = TDMT_SYNC_CONFIG_CHANGE;
   rpcMsg.info.noResp = 1;
@@ -189,58 +263,59 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
   rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
   snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", newconfig);
   taosMemoryFree(newconfig);
-  ret = syncPropose(rid, &rpcMsg, false);
+  ret = syncNodePropose(pSyncNode, &rpcMsg, false);
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
   return ret;
 }
 
 int32_t syncLeaderTransfer(int64_t rid) {
-  int32_t ret = 0;
+  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
+  if (pSyncNode == NULL) {
+    return TAOS_SYNC_OTHER_ERROR;
+  }
+  ASSERT(rid == pSyncNode->rid);
+
+  if (pSyncNode->peersNum == 0) {
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+    return TAOS_SYNC_OTHER_ERROR;
+  }
+
+  SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
 
+  int32_t ret = syncLeaderTransferTo(rid, newLeader);
   return ret;
 }
 
 int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return false;
+    return TAOS_SYNC_OTHER_ERROR;
   }
-  assert(rid == pSyncNode->rid);
+  ASSERT(rid == pSyncNode->rid);
   int32_t ret = 0;
 
   if (pSyncNode->replicaNum == 1) {
-    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
     sError("only one replica, cannot drop leader");
+    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
     return TAOS_SYNC_ONLY_ONE_REPLICA;
   }
 
   SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId);
   pMsg->newLeaderId.addr = syncUtilAddr2U64(newLeader.nodeFqdn, newLeader.nodePort);
   pMsg->newLeaderId.vgId = pSyncNode->vgId;
+  pMsg->newNodeInfo = newLeader;
   ASSERT(pMsg != NULL);
   SRpcMsg rpcMsg = {0};
   syncLeaderTransfer2RpcMsg(pMsg, &rpcMsg);
   syncLeaderTransferDestroy(pMsg);
 
-  ret = syncPropose(rid, &rpcMsg, false);
-
+  ret = syncNodePropose(pSyncNode, &rpcMsg, false);
   taosReleaseRef(tsNodeRefId, pSyncNode->rid);
   return ret;
 }
 
-int32_t syncReconfigRaw(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
-  int32_t ret = 0;
-  char*   newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
-
-  pRpcMsg->msgType = TDMT_SYNC_CONFIG_CHANGE;
-  pRpcMsg->info.noResp = 1;
-  pRpcMsg->contLen = strlen(newconfig) + 1;
-  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
-  snprintf(pRpcMsg->pCont, pRpcMsg->contLen, "%s", newconfig);
-  taosMemoryFree(newconfig);
-
-  return ret;
-}
-
 bool syncCanLeaderTransfer(int64_t rid) {
   SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
@@ -273,8 +348,6 @@ bool syncCanLeaderTransfer(int64_t rid) {
   return matchOK;
 }
 
-int32_t syncGiveUpLeader(int64_t rid) { return 0; }
-
 int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
   int32_t ret = syncPropose(rid, pMsg, isWeak);
   return ret;
@@ -469,16 +542,26 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
 
   SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
   if (pSyncNode == NULL) {
-    return TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    return TAOS_SYNC_OTHER_ERROR;
   }
   assert(rid == pSyncNode->rid);
+  sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
+  ret = syncNodePropose(pSyncNode, pMsg, isWeak);
+
+  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
+  return ret;
+}
+
+int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) {
+  int32_t ret = TAOS_SYNC_PROPOSE_SUCCESS;
+  sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
     SRespStub stub;
     stub.createTime = taosGetTimestampMs();
     stub.rpcMsg = *pMsg;
     uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
-    sDebug("vgId:%d, sync event propose, type:%s seq:%" PRIu64 " handle:%p", pSyncNode->vgId, TMSG_INFO(pMsg->msgType),
+    sDebug("vgId:%d sync event propose, type:%s seq:%" PRIu64 " handle:%p", pSyncNode->vgId, TMSG_INFO(pMsg->msgType),
            seqNum, pMsg->info.handle);
 
     SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, seqNum, isWeak, pSyncNode->vgId);
@@ -488,16 +571,14 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
     if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
       ret = TAOS_SYNC_PROPOSE_SUCCESS;
     } else {
-      sTrace("syncPropose pSyncNode->FpEqMsg is NULL");
+      sError("syncPropose pSyncNode->FpEqMsg is NULL");
     }
     syncClientRequestDestroy(pSyncMsg);
   } else {
-    sDebug("vgId:%d, failed to propose since not leader, type:%s handle:%p %s", pSyncNode->vgId,
-           TMSG_INFO(pMsg->msgType), pMsg->info.handle, syncUtilState2String(pSyncNode->state));
+    sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state));
     ret = TAOS_SYNC_PROPOSE_NOT_LEADER;
   }
 
-  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
   return ret;
 }
 
@@ -1162,8 +1243,14 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
   int32_t oldReplicaNum = pSyncNode->replicaNum;
   SRaftId oldReplicasId[TSDB_MAX_REPLICA];
   memcpy(oldReplicasId, pSyncNode->replicasId, sizeof(oldReplicasId));
-  SSyncSnapshotSender*   oldSenders[TSDB_MAX_REPLICA];
-  memcpy(oldSenders, pSyncNode->senders, sizeof(oldSenders));
+  SSyncSnapshotSender* oldSenders[TSDB_MAX_REPLICA];
+  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
+    oldSenders[i] = (pSyncNode->senders)[i];
+    sDebug("vgId:%d sync event save senders %d, %p", pSyncNode->vgId, i, oldSenders[i]);
+    if (gRaftDetailLog) {
+      ;
+    }
+  }
 
   // init internal
   pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
@@ -1195,24 +1282,51 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
 
   pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);
 
-  // reset snapshot senders, memory leak
+  // reset snapshot senders
+
+  // clear new
   for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
     (pSyncNode->senders)[i] = NULL;
   }
+
+  // reset new
   for (int i = 0; i < pSyncNode->replicaNum; ++i) {
+    // reset sender
+    bool reset = false;
     for (int j = 0; j < TSDB_MAX_REPLICA; ++j) {
       if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j])) {
-        char host[128];
+        char     host[128];
         uint16_t port;
         syncUtilU642Addr((pSyncNode->replicasId)[i].addr, host, sizeof(host), &port);
-        sDebug("vgId:%d sync event reset sender for %lu, %s:%d", pSyncNode->vgId, (pSyncNode->replicasId)[i].addr, host, port);
+        sDebug("vgId:%d sync event reset sender for %lu, newIndex:%d, %s:%d, %p", pSyncNode->vgId,
+               (pSyncNode->replicasId)[i].addr, i, host, port, oldSenders[j]);
         (pSyncNode->senders)[i] = oldSenders[j];
+        oldSenders[j] = NULL;
+        reset = true;
+
+        // reset replicaIndex
+        int32_t oldreplicaIndex = (pSyncNode->senders)[i]->replicaIndex;
+        (pSyncNode->senders)[i]->replicaIndex = i;
+        sDebug("vgId:%d sync event udpate replicaIndex from %d to %d, %s:%d, %p, reset:%d", pSyncNode->vgId,
+               oldreplicaIndex, i, host, port, (pSyncNode->senders)[i], reset);
       }
     }
   }
+
+  // create new
   for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
     if ((pSyncNode->senders)[i] == NULL) {
       (pSyncNode->senders)[i] = snapshotSenderCreate(pSyncNode, i);
+      sDebug("vgId:%d sync event create new sender %p replicaIndex:%d", pSyncNode->vgId, (pSyncNode->senders)[i], i);
+    }
+  }
+
+  // free old
+  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
+    if (oldSenders[i] != NULL) {
+      snapshotSenderDestroy(oldSenders[i]);
+      sDebug("vgId:%d sync event delete old sender %p replicaIndex:%d", pSyncNode->vgId, oldSenders[i], i);
+      oldSenders[i] = NULL;
     }
   }
 
@@ -1272,8 +1386,8 @@ void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
 }
 
 void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
-  sDebug("vgId:%d sync event become follower, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy,
-         debugStr);
+  sDebug("vgId:%d sync event become follower, isStandBy:%d, replicaNum:%d, %s", pSyncNode->vgId,
+         pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, debugStr);
 
   // maybe clear leader cache
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
@@ -1307,8 +1421,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
 //     /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
 //
 void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
-  sDebug("vgId:%d sync event become leader, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy,
-         debugStr);
+  sDebug("vgId:%d sync event become leader, isStandBy:%d, replicaNum:%d %s", pSyncNode->vgId,
+         pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, debugStr);
 
   // state change
   pSyncNode->state = TAOS_SYNC_STATE_LEADER;
@@ -1857,10 +1971,52 @@ const char* syncStr(ESyncState state) {
 }
 
 static int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
-  SyncLeaderTransfer* pSyncLeaderTransfer;
-  if (syncUtilSameId(&(pSyncLeaderTransfer->newLeaderId), &(ths->myRaftId))) {
+  SyncLeaderTransfer* pSyncLeaderTransfer = syncLeaderTransferFromRpcMsg2(pRpcMsg);
+
+  /*
+    char     host[128];
+    uint16_t port;
+    syncUtilU642Addr(pSyncLeaderTransfer->newLeaderId.addr, host, sizeof(host), &port);
+    sDebug("vgId:%d sync event, maybe leader transfer to %s:%d %lu", ths->vgId, host, port,
+           pSyncLeaderTransfer->newLeaderId.addr);
+  */
+
+  sDebug("vgId:%d sync event, begin leader transfer", ths->vgId);
+
+  if (strcmp(pSyncLeaderTransfer->newNodeInfo.nodeFqdn, ths->myNodeInfo.nodeFqdn) == 0 &&
+      pSyncLeaderTransfer->newNodeInfo.nodePort == ths->myNodeInfo.nodePort) {
+    sDebug("vgId:%d sync event, maybe leader transfer to %s:%d %lu", ths->vgId,
+           pSyncLeaderTransfer->newNodeInfo.nodeFqdn, pSyncLeaderTransfer->newNodeInfo.nodePort,
+           pSyncLeaderTransfer->newLeaderId.addr);
+
+    // reset elect timer now!
+    int32_t electMS = 1;
+    int32_t ret = syncNodeRestartElectTimer(ths, electMS);
+    ASSERT(ret == 0);
   }
 
+  /*
+    if (syncUtilSameId(&(pSyncLeaderTransfer->newLeaderId), &(ths->myRaftId))) {
+      // reset elect timer now!
+      int32_t electMS = 1;
+      int32_t ret = syncNodeRestartElectTimer(ths, electMS);
+      ASSERT(ret == 0);
+    }
+  */
+  if (ths->pFsm->FpLeaderTransferCb != NULL) {
+    SFsmCbMeta cbMeta;
+    cbMeta.code = 0;
+    cbMeta.currentTerm = ths->pRaftStore->currentTerm;
+    cbMeta.flag = 0;
+    cbMeta.index = pEntry->index;
+    cbMeta.isWeak = pEntry->isWeak;
+    cbMeta.seqNum = pEntry->seqNum;
+    cbMeta.state = ths->state;
+    cbMeta.term = pEntry->term;
+    ths->pFsm->FpLeaderTransferCb(ths->pFsm, pRpcMsg, cbMeta);
+  }
+
+  syncLeaderTransferDestroy(pSyncLeaderTransfer);
   return 0;
 }
 
@@ -1884,26 +2040,31 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE
 
   bool isDrop;
 
-  //if (IamInNew || (!IamInNew && ths->state != TAOS_SYNC_STATE_LEADER)) {
+  // if (IamInNew || (!IamInNew && ths->state != TAOS_SYNC_STATE_LEADER)) {
   if (IamInNew) {
     syncNodeUpdateConfig(ths, &newSyncCfg, pEntry->index, &isDrop);
 
     // change isStandBy to normal
     if (!isDrop) {
+      char tmpbuf[128];
+      snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d", oldSyncCfg.replicaNum, newSyncCfg.replicaNum);
       if (ths->state == TAOS_SYNC_STATE_LEADER) {
-        syncNodeBecomeLeader(ths, "config change");
+        syncNodeBecomeLeader(ths, tmpbuf);
       } else {
-        syncNodeBecomeFollower(ths, "config change");
+        syncNodeBecomeFollower(ths, tmpbuf);
       }
     }
   } else {
-    syncNodeBecomeFollower(ths, "config change2");
+    char tmpbuf[128];
+    snprintf(tmpbuf, sizeof(tmpbuf), "config change2 from %d to %d", oldSyncCfg.replicaNum, newSyncCfg.replicaNum);
+    syncNodeBecomeFollower(ths, tmpbuf);
   }
-  
+
   if (gRaftDetailLog) {
     char* sOld = syncCfg2Str(&oldSyncCfg);
     char* sNew = syncCfg2Str(&newSyncCfg);
-    sInfo("==config change== 0x11 old:%s new:%s isDrop:%d index:%ld \n", sOld, sNew, isDrop, pEntry->index);
+    sInfo("==config change== 0x11 old:%s new:%s isDrop:%d index:%ld IamInNew:%d \n", sOld, sNew, isDrop, pEntry->index,
+          IamInNew);
     taosMemoryFree(sOld);
     taosMemoryFree(sNew);
   }
@@ -1965,7 +2126,7 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
           ASSERT(code == 0);
         }
 
-        // config change
+        // leader transfer
         if (pEntry->originalRpcType == TDMT_SYNC_LEADER_TRANSFER) {
           code = syncDoLeaderTransfer(ths, &rpcMsg, pEntry);
           ASSERT(code == 0);

source/libs/sync/src/syncSnapshot.c

@@ -393,7 +393,7 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
 
 char *snapshotSender2Str(SSyncSnapshotSender *pSender) {
   cJSON *pJson = snapshotSender2Json(pSender);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -514,7 +514,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
     cJSON_AddStringToObject(pFromId, "addr", u64buf);
     {
       uint64_t u64 = pReceiver->fromId.addr;
-      cJSON *  pTmp = pFromId;
+      cJSON   *pTmp = pFromId;
       char     host[128] = {0};
       uint16_t port;
       syncUtilU642Addr(u64, host, sizeof(host), &port);
@@ -538,7 +538,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
 
 char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
   cJSON *pJson = snapshotReceiver2Json(pReceiver);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -588,6 +588,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
 
         // maybe update lastconfig
         if (pMsg->lastConfigIndex >= SYNC_INDEX_BEGIN) {
+          int32_t oldReplicaNum = pSyncNode->replicaNum;
+
           // update new config myIndex
           bool     IamInNew = false;
           SSyncCfg newSyncCfg = pMsg->lastConfig;
@@ -614,10 +616,12 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
 
           // change isStandBy to normal
           if (!isDrop) {
+            char tmpbuf[128];
+            snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d", oldReplicaNum, newSyncCfg.replicaNum);
             if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
-              syncNodeBecomeLeader(pSyncNode, "config change");
+              syncNodeBecomeLeader(pSyncNode, tmpbuf);
             } else {
-              syncNodeBecomeFollower(pSyncNode, "config change");
+              syncNodeBecomeFollower(pSyncNode, tmpbuf);
             }
           }
         }

source/libs/sync/test/CMakeLists.txt

 add_executable(syncTest "")
+add_executable(syncRaftIdCheck "")
 add_executable(syncEnvTest "")
 add_executable(syncPingTimerTest "")
 add_executable(syncIOTickQTest "")
@@ -54,6 +55,10 @@ target_sources(syncTest
     PRIVATE
     "syncTest.cpp"
 )
+target_sources(syncRaftIdCheck
+    PRIVATE
+    "syncRaftIdCheck.cpp"
+)
 target_sources(syncEnvTest
     PRIVATE
     "syncEnvTest.cpp"
@@ -257,6 +262,11 @@ target_include_directories(syncTest
     "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncRaftIdCheck
+    PUBLIC
+    "${TD_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 target_include_directories(syncEnvTest
     PUBLIC
     "${TD_SOURCE_DIR}/include/libs/sync"
@@ -508,6 +518,10 @@ target_link_libraries(syncTest
     sync
     gtest_main
 )
+target_link_libraries(syncRaftIdCheck
+    sync
+    gtest_main
+)
 target_link_libraries(syncEnvTest
     sync
     gtest_main

source/libs/sync/test/syncRaftIdCheck.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncUtil.h"
+
+void usage(char* exe) {
+  printf("Usage: %s host port \n", exe);
+  printf("Usage: %s u64 \n", exe);
+}
+
+int main(int argc, char** argv) {
+  if (argc == 2) {
+    uint64_t u64 = atoll(argv[1]);
+    char     host[128];
+    uint16_t port;
+    syncUtilU642Addr(u64, host, sizeof(host), &port);
+    printf("%lu -> %s:%d \n", u64, host, port);
+
+  } else if (argc == 3) {
+    uint64_t u64;
+    char*    host = argv[1];
+    uint16_t port = atoi(argv[2]);
+    u64 = syncUtilAddr2U64(host, port);
+    printf("%s:%d -> %lu \n", host, port, u64);
+  } else {
+    usage(argv[0]);
+    exit(-1);
+  }
+
+  return 0;
+}

source/libs/sync/test/syncTestTool.cpp

@@ -153,6 +153,16 @@ void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMe
   taosMemoryFree(s);
 }
 
+void LeaderTransferCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta) {
+  char logBuf[256] = {0};
+  snprintf(logBuf, sizeof(logBuf),
+           "==callback== ==LeaderTransferCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s, flag:%lu, term:%lu "
+           "currentTerm:%lu \n",
+           pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state),
+           cbMeta.flag, cbMeta.term, cbMeta.currentTerm);
+  syncRpcMsgLog2(logBuf, (SRpcMsg*)pMsg);
+}
+
 SSyncFSM* createFsm() {
   SSyncFSM* pFsm = (SSyncFSM*)taosMemoryMalloc(sizeof(SSyncFSM));
   memset(pFsm, 0, sizeof(*pFsm));
@@ -172,6 +182,8 @@ SSyncFSM* createFsm() {
   pFsm->FpSnapshotStopWrite = SnapshotStopWrite;
   pFsm->FpSnapshotDoWrite = SnapshotDoWrite;
 
+  pFsm->FpLeaderTransferCb = LeaderTransferCb;
+
   return pFsm;
 }
 
@@ -277,7 +289,8 @@ void usage(char* exe) {
   printf(
       "usage: %s  replicaNum(1-5)  myIndex(0-..)  enableSnapshot(0/1)  lastApplyIndex(>=-1)  lastApplyTerm(>=0)  "
       "writeRecordNum(>=0)  "
-      "isStandBy(0/1)  isConfigChange(0-5)  iterTimes(>=0)  finishLastApplyIndex(>=-1)  finishLastApplyTerm(>=0) \n",
+      "isStandBy(0/1)  isConfigChange(0-5)  iterTimes(>=0)  finishLastApplyIndex(>=-1)  finishLastApplyTerm(>=0) "
+      "leaderTransfer(0/1) \n",
       exe);
 }
 
@@ -294,9 +307,9 @@ SRpcMsg* createRpcMsg(int i, int count, int myIndex) {
 int main(int argc, char** argv) {
   sprintf(tsTempDir, "%s", ".");
   tsAsyncLog = 0;
-  sDebugFlag = DEBUG_SCREEN + DEBUG_FILE + DEBUG_TRACE + DEBUG_INFO + DEBUG_ERROR;
+  sDebugFlag = DEBUG_SCREEN + DEBUG_FILE + DEBUG_TRACE + DEBUG_INFO + DEBUG_ERROR + DEBUG_DEBUG;
 
-  if (argc != 12) {
+  if (argc != 13) {
     usage(argv[0]);
     exit(-1);
   }
@@ -312,12 +325,14 @@ int main(int argc, char** argv) {
   int32_t iterTimes = atoi(argv[9]);
   int32_t finishLastApplyIndex = atoi(argv[10]);
   int32_t finishLastApplyTerm = atoi(argv[11]);
+  int32_t leaderTransfer = atoi(argv[12]);
 
-  sTrace(
+  sInfo(
       "args: replicaNum:%d, myIndex:%d, enableSnapshot:%d, lastApplyIndex:%d, lastApplyTerm:%d, writeRecordNum:%d, "
-      "isStandBy:%d, isConfigChange:%d, iterTimes:%d, finishLastApplyIndex:%d, finishLastApplyTerm:%d",
+      "isStandBy:%d, isConfigChange:%d, iterTimes:%d, finishLastApplyIndex:%d, finishLastApplyTerm:%d, "
+      "leaderTransfer:%d",
       replicaNum, myIndex, enableSnapshot, lastApplyIndex, lastApplyTerm, writeRecordNum, isStandBy, isConfigChange,
-      iterTimes, finishLastApplyIndex, finishLastApplyTerm);
+      iterTimes, finishLastApplyIndex, finishLastApplyTerm, leaderTransfer);
 
   // check parameter
   assert(replicaNum >= 1 && replicaNum <= 5);
@@ -363,24 +378,31 @@ int main(int argc, char** argv) {
 
   //---------------------------
   int32_t alreadySend = 0;
+  int32_t leaderTransferWait = 0;
   while (1) {
     char* simpleStr = syncNode2SimpleStr(pSyncNode);
 
+    leaderTransferWait++;
+    if (leaderTransferWait == 7) {
+      sTrace("begin leader transfer ...");
+      int32_t ret = syncLeaderTransfer(rid);
+    }
+
     if (alreadySend < writeRecordNum) {
       SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
       int32_t  ret = syncPropose(rid, pRpcMsg, false);
       if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
-        sTrace("%s value%d write not leader", simpleStr, alreadySend);
+        sTrace("%s value%d write not leader, leaderTransferWait:%d", simpleStr, alreadySend, leaderTransferWait);
       } else {
         assert(ret == 0);
-        sTrace("%s value%d write ok", simpleStr, alreadySend);
+        sTrace("%s value%d write ok, leaderTransferWait:%d", simpleStr, alreadySend, leaderTransferWait);
       }
       alreadySend++;
 
       rpcFreeCont(pRpcMsg->pCont);
       taosMemoryFree(pRpcMsg);
     } else {
-      sTrace("%s", simpleStr);
+      sTrace("%s, leaderTransferWait:%d", simpleStr, leaderTransferWait);
     }
 
     taosMsleep(1000);

source/os/src/osLz4.c

@@ -45,13 +45,13 @@ int32_t BUILDIN_CLZL(uint64_t val) {
 #else
   _MyBitScanReverse64(&r, val);
 #endif
-  return (int)(r >> 3);
+  return (int)(63 - r);
 }
 
 int32_t BUILDIN_CLZ(uint32_t val) {
   unsigned long r = 0;
   _BitScanReverse(&r, val);
-  return (int)(r >> 3);
+  return (int)(31 - r);
 }
 
 int32_t BUILDIN_CTZL(uint64_t val) {
@@ -61,13 +61,13 @@ int32_t BUILDIN_CTZL(uint64_t val) {
 #else
   _MyBitScanForward64(&r, val);
 #endif
-  return (int)(r >> 3);
+  return (int)(r);
 }
 
 int32_t BUILDIN_CTZ(uint32_t val) {
   unsigned long r = 0;
   _BitScanForward(&r, val);
-  return (int)(r >> 3);
+  return (int)(r);
 }
 
 #endif

source/util/src/tarray.c

@@ -458,6 +458,9 @@ static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void
 }
 
 SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy) {
+  if (NULL == pSrc) {
+    return NULL;
+  }
   ASSERT(pSrc->elemSize == sizeof(void*));
   SArray* pArray = taosArrayInit(pSrc->size, sizeof(void*));
   for (int32_t i = 0; i < pSrc->size; i++) {

tests/script/tsim/stable/tag_filter.sim

@@ -73,5 +73,80 @@ if $rows != 6 then
 endi
 
 
+print ========== prepare stbBin and ctbBin
+sql create table db.stbBin (ts timestamp, c1 int, c2 binary(4)) tags(t1 binary(16))
 
+
+sql create table db.ctbBin using db.stbBin tags("a")
+sql insert into db.ctbBin values(now, 1, "2")
+
+sql create table db.ctbBin1 using db.stbBin tags("b")
+sql insert into db.ctbBin1 values(now, 2, "2")
+
+sql create table db.ctbBin2 using db.stbBin tags("c")
+sql insert into db.ctbBin2 values(now, 3, "2")
+
+sql create table db.ctbBin3 using db.stbBin tags("d")
+sql insert into db.ctbBin3 values(now, 4, "2")
+
+sql select * from db.stbBin where t1 = "a" 
+if $rows != 1 then 
+  return -1
+endi
+
+
+sql select * from db.stbBin where t1 < "a" 
+if $rows != 0 then 
+  return -=1
+endi
+
+sql select * from db.stbBin where t1 < "b" 
+if $rows != 1 then 
+  return -1
+endi
+
+
+sql select * from db.stbBin where t1 between "a" and "e" 
+if $rows != 4 then 
+  return -1
+endi
+
+
+print ========== prepare stbNc and ctbNc
+sql create table db.stbNc (ts timestamp, c1 int, c2 binary(4)) tags(t1 nchar(16))
+
+
+sql create table db.ctbNc using db.stbNc tags("a")
+sql insert into db.ctbNc values(now, 1, "2")
+
+sql create table db.ctbNc1 using db.stbNc tags("b")
+sql insert into db.ctbNc1 values(now, 2, "2")
+
+sql create table db.ctbNc2 using db.stbNc tags("c")
+sql insert into db.ctbNc2 values(now, 3, "2")
+
+sql create table db.ctbNc3 using db.stbNc tags("d")
+sql insert into db.ctbNc3 values(now, 4, "2")
+
+sql select * from db.stbNc where t1 = "a" 
+if $rows != 1 then 
+  return -1
+endi
+
+
+sql select * from db.stbNc where t1 < "a" 
+if $rows != 0 then 
+  return -=1
+endi
+
+sql select * from db.stbNc where t1 < "b" 
+if $rows != 1 then 
+  return -1
+endi
+
+
+sql select * from db.stbNc where t1 between "a" and "e" 
+if $rows != 4 then 
+  return -1
+endi
 system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/system-test/7-tmq/subscribeDb1.py

@@ -437,12 +437,12 @@ class TDTestCase:
         event.wait()
 
         tdLog.info("start consume processor")
-        pollDelay = 100
+        pollDelay = 20
         showMsg   = 1
         showRow   = 1
         self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
 
-        time.sleep(5)
+        time.sleep(3)
         tdLog.info("pkill consume processor")
         if (platform.system().lower() == 'windows'):
             os.system("TASKKILL /F /IM tmq_sim.exe")
@@ -465,7 +465,7 @@ class TDTestCase:
             totalConsumeRows += resultList[i]
         
         if totalConsumeRows >= expectrowcnt or totalConsumeRows <= 0:
-            tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt))
+            tdLog.info("act consume rows: %d, expect consume rows between %d and 0"%(totalConsumeRows, expectrowcnt))
             tdLog.exit("tmq consume rows error!")
 
         time.sleep(15)

tests/system-test/fulltest.sh

@@ -104,7 +104,7 @@ python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py
 python3 ./test.py -f 7-tmq/basic5.py
 python3 ./test.py -f 7-tmq/subscribeDb.py
 python3 ./test.py -f 7-tmq/subscribeDb0.py
-# python3 ./test.py -f 7-tmq/subscribeDb1.py
+python3 ./test.py -f 7-tmq/subscribeDb1.py
 python3 ./test.py -f 7-tmq/subscribeStb.py
 python3 ./test.py -f 7-tmq/subscribeStb0.py
 python3 ./test.py -f 7-tmq/subscribeStb1.py