Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/row_refact

include/common/tmsgcb.h

@@ -38,7 +38,7 @@ typedef enum {
   QUEUE_MAX,
 } EQueueType;
 
-typedef int32_t (*PutToQueueFp)(void* pMgmt, SRpcMsg* pMsg);
+typedef int32_t (*PutToQueueFp)(void* pMgmt, EQueueType qtype, SRpcMsg* pMsg);
 typedef int32_t (*GetQueueSizeFp)(void* pMgmt, int32_t vgId, EQueueType qtype);
 typedef int32_t (*SendReqFp)(const SEpSet* pEpSet, SRpcMsg* pMsg);
 typedef void (*SendRspFp)(SRpcMsg* pMsg);
@@ -50,7 +50,7 @@ typedef void (*ReportStartup)(const char* name, const char* desc);
 typedef struct {
   void*                   mgmt;
   void*                   clientRpc;
-  PutToQueueFp            queueFps[QUEUE_MAX];
+  PutToQueueFp            putToQueueFp;
   GetQueueSizeFp          qsizeFp;
   SendReqFp               sendReqFp;
   SendRspFp               sendRspFp;

include/common/tmsgdef.h

@@ -70,106 +70,102 @@ enum {
   // Requests handled by DNODE
   TD_NEW_MSG_SEG(TDMT_DND_MSG)
   TD_DEF_MSG_TYPE(TDMT_DND_CREATE_MNODE, "dnode-create-mnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_ALTER_MNODE, "dnode-alter-mnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_DROP_MNODE, "dnode-drop-mnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_CREATE_QNODE, "dnode-create-qnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_ALTER_QNODE, "dnode-alter-qnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_DROP_QNODE, "dnode-drop-qnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_CREATE_SNODE, "dnode-create-snode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_ALTER_SNODE, "dnode-alter-snode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_DROP_SNODE, "dnode-drop-snode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_CREATE_BNODE, "dnode-create-bnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_ALTER_BNODE, "dnode-alter-bnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_DROP_BNODE, "dnode-drop-bnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "dnode-create-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "dnode-server-status", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "dnode-net-test", NULL, NULL)
-
+  TD_DEF_MSG_TYPE(TDMT_DND_CREATE_VNODE, "create-vnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "drop-vnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "server-status", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "net-test", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "config-dnode", NULL, NULL)
+  
   // Requests handled by MNODE
   TD_NEW_MSG_SEG(TDMT_MND_MSG)
-  TD_DEF_MSG_TYPE(TDMT_MND_CONNECT, "mnode-connect", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_ACCT, "mnode-create-acct", NULL, NULL)	
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_ACCT, "mnode-alter-acct", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_ACCT, "mnode-drop-acct", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_USER, "mnode-create-user", NULL, NULL)	
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_USER, "mnode-alter-user", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_USER, "mnode-drop-user", NULL, NULL) 
-  TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_AUTH, "mnode-get-user-auth", NULL, NULL) 
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DNODE, "mnode-create-dnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CONFIG_DNODE, "mnode-config-dnode", NULL, NULL) 
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DNODE, "mnode-alter-dnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_DNODE, "mnode-drop-dnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "mnode-create-mnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "mnode-alter-mnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "mnode-drop-mnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "mnode-create-qnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "mnode-alter-qnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "mnode-drop-qnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SNODE, "mnode-create-snode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_SNODE, "mnode-alter-snode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_SNODE, "mnode-drop-snode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_BNODE, "mnode-create-bnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_BNODE, "mnode-alter-bnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_BNODE, "mnode-drop-bnode", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DB, "mnode-create-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_DB, "mnode-drop-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_USE_DB, "mnode-use-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DB, "mnode-alter-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_SYNC_DB, "mnode-sync-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_COMPACT_DB, "mnode-compact-db", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_FUNC, "mnode-create-func", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_FUNC, "mnode-retrieve-func", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_FUNC, "mnode-drop-func", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STB, "mnode-create-stb", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STB, "mnode-alter-stb", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_STB, "mnode-drop-stb", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SMA, "mnode-create-sma", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_SMA, "mnode-drop-sma", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_TABLE_META, "mnode-table-meta", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "mnode-vgroup-list", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_QNODE_LIST, "mnode-qnode-list", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "mnode-kill-query", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "mnode-kill-conn", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", SClientHbBatchReq, SClientHbBatchRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "mnode-show", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_SYSTABLE_RETRIEVE, "mnode-systable-retrieve", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "mnode-status", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "mnode-trans-tmr", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "mnode-kill-trans", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "mnode-telem-tmr", SMTimerReq, SMTimerReq)
-  TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "mnode-grant", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "mnode-auth", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "mnode-create-topic", SMCreateTopicReq, SMCreateTopicRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMqConsumerLostMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "mnode-mq-consumer-recover", SMqConsumerRecoverMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "mnode-mq-drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "mnode-create-stream", SCMCreateStreamReq, SCMCreateStreamRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "mnode-alter-stream", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_STREAM, "mnode-drop-stream", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_INDEX, "mnode-create-index", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "mnode-drop-index", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_CFG, "mnode-get-db-cfg", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "mnode-get-index", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "mnode-apply-msg", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CONNECT, "connect", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_ACCT, "create-acct", NULL, NULL)	
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_ACCT, "alter-acct", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_ACCT, "drop-acct", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_USER, "create-user", NULL, NULL)	
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_USER, "alter-user", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_USER, "drop-user", NULL, NULL) 
+  TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_AUTH, "get-user-auth", NULL, NULL) 
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DNODE, "create-dnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CONFIG_DNODE, "config-dnode", NULL, NULL) 
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DNODE, "alter-dnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_DNODE, "drop-dnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "create-mnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "alter-mnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "drop-mnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "create-qnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "alter-qnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "drop-qnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SNODE, "create-snode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_SNODE, "alter-snode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_SNODE, "drop-snode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_BNODE, "create-bnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_BNODE, "alter-bnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_BNODE, "drop-bnode", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_DB, "create-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_DB, "drop-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_USE_DB, "use-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_DB, "alter-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SYNC_DB, "sync-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_COMPACT_DB, "compact-db", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_FUNC, "create-func", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_FUNC, "retrieve-func", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_FUNC, "drop-func", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STB, "create-stb", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STB, "alter-stb", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_STB, "drop-stb", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_SMA, "create-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_SMA, "drop-sma", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TABLE_META, "table-meta", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "vgroup-list", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_QNODE_LIST, "qnode-list", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "kill-query", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "kill-conn", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "heartbeat", SClientHbBatchReq, SClientHbBatchRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "show", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SYSTABLE_RETRIEVE, "systable-retrieve", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "status", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "kill-trans", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq)
+  TD_DEF_MSG_TYPE(TDMT_MND_GRANT, "grant", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_AUTH, "auth", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "alter-topic", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "drop-topic", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "subscribe", SCMSubscribeReq, SCMSubscribeRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mq-ask-ep", SMqAskEpReq, SMqAskEpRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mq-tmr", SMTimerReq, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mq-consumer-lost", SMqConsumerLostMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "mq-consumer-recover", SMqConsumerRecoverMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mq-do-rebalance", SMqDoRebalanceMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "mq-drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "create-stream", SCMCreateStreamReq, SCMCreateStreamRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "alter-stream", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_STREAM, "drop-stream", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_INDEX, "create-index", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "drop-index", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_CFG, "get-db-cfg", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "get-index", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_APPLY_MSG, "apply-msg", NULL, NULL)
 
   // Requests handled by VNODE
   TD_NEW_MSG_SEG(TDMT_VND_MSG)
-  TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "vnode-submit", SSubmitReq, SSubmitRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_QUERY, "vnode-query", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_FETCH, "vnode-fetch", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TABLE, "vnode-create-table", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TABLE, "vnode-alter-table", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_DROP_TABLE, "vnode-drop-table", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_UPDATE_TAG_VAL, "vnode-update-tag-val", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT, "submit", SSubmitReq, SSubmitRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_QUERY, "query", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_FETCH, "fetch", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TABLE, "create-table", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TABLE, "alter-table", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_DROP_TABLE, "drop-table", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_UPDATE_TAG_VAL, "update-tag-val", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TABLE_META, "vnode-table-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TABLES_META, "vnode-tables-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_STB, "vnode-create-stb", SVCreateStbReq, NULL)

include/common/ttypes.h

@@ -30,7 +30,7 @@ typedef uint64_t TDRowVerT;
 typedef int16_t  col_id_t;
 typedef int8_t   col_type_t;
 typedef int32_t  col_bytes_t;
-typedef int32_t schema_ver_t;
+typedef int32_t  schema_ver_t;
 typedef int32_t  func_id_t;
 
 #pragma pack(push, 1)
@@ -49,8 +49,9 @@ typedef struct {
 #define varDataCopy(dst, v)    memcpy((dst), (void *)(v), varDataTLen(v))
 #define varDataLenByData(v)    (*(VarDataLenT *)(((char *)(v)) - VARSTR_HEADER_SIZE))
 #define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT)(_len))
-#define IS_VAR_DATA_TYPE(t)    (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
-#define IS_STR_DATA_TYPE(t)    (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
+#define IS_VAR_DATA_TYPE(t) \
+  (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
+#define IS_STR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
 
 #define varDataNetLen(v)  (htons(((VarDataLenT *)(v))[0]))
 #define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v))
@@ -142,52 +143,56 @@ typedef struct {
     }                                          \
   } while (0)
 
-#define NUM_TO_STRING(_inputType, _input, _outputBytes, _output)     \
-  do {                                         \
-    switch (_inputType) {                      \
-      case TSDB_DATA_TYPE_TINYINT:             \
-        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int8_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_UTINYINT:            \
-        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint8_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_SMALLINT:            \
-        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int16_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_USMALLINT:           \
-        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint16_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_TIMESTAMP:           \
-      case TSDB_DATA_TYPE_BIGINT:              \
-        snprintf(_output, (int32_t)(_outputBytes), "%" PRId64, *(int64_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_UBIGINT:             \
-        snprintf(_output, (int32_t)(_outputBytes), "%" PRIu64, *(uint64_t *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_FLOAT:               \
-        snprintf(_output, (int32_t)(_outputBytes), "%f", *(float *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_DOUBLE:              \
-        snprintf(_output, (int32_t)(_outputBytes), "%f", *(double *)(_input));     \
-        break;                                 \
-      case TSDB_DATA_TYPE_UINT:                \
-        snprintf(_output, (int32_t)(_outputBytes), "%u", *(uint32_t *)(_input));     \
-        break;                                 \
-      default:                                 \
-        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int32_t *)(_input));     \
-        break;                                 \
-    }                                          \
+#define NUM_TO_STRING(_inputType, _input, _outputBytes, _output)                       \
+  do {                                                                                 \
+    switch (_inputType) {                                                              \
+      case TSDB_DATA_TYPE_TINYINT:                                                     \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int8_t *)(_input));         \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_UTINYINT:                                                    \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint8_t *)(_input));        \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_SMALLINT:                                                    \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int16_t *)(_input));        \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_USMALLINT:                                                   \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(uint16_t *)(_input));       \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_TIMESTAMP:                                                   \
+      case TSDB_DATA_TYPE_BIGINT:                                                      \
+        snprintf(_output, (int32_t)(_outputBytes), "%" PRId64, *(int64_t *)(_input));  \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_UBIGINT:                                                     \
+        snprintf(_output, (int32_t)(_outputBytes), "%" PRIu64, *(uint64_t *)(_input)); \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_FLOAT:                                                       \
+        snprintf(_output, (int32_t)(_outputBytes), "%f", *(float *)(_input));          \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_DOUBLE:                                                      \
+        snprintf(_output, (int32_t)(_outputBytes), "%f", *(double *)(_input));         \
+        break;                                                                         \
+      case TSDB_DATA_TYPE_UINT:                                                        \
+        snprintf(_output, (int32_t)(_outputBytes), "%u", *(uint32_t *)(_input));       \
+        break;                                                                         \
+      default:                                                                         \
+        snprintf(_output, (int32_t)(_outputBytes), "%d", *(int32_t *)(_input));        \
+        break;                                                                         \
+    }                                                                                  \
   } while (0)
 
+
 //TODO: use varchar(0) to represent NULL type
+#define IS_VAR_NULL_TYPE(_t, _b)     ((_t) == TSDB_DATA_TYPE_VARCHAR && (_b) == 0)
 #define IS_NULL_TYPE(_t)             ((_t) == TSDB_DATA_TYPE_NULL)
+
 #define IS_SIGNED_NUMERIC_TYPE(_t)   ((_t) >= TSDB_DATA_TYPE_TINYINT && (_t) <= TSDB_DATA_TYPE_BIGINT)
 #define IS_UNSIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_UTINYINT && (_t) <= TSDB_DATA_TYPE_UBIGINT)
 #define IS_FLOAT_TYPE(_t)            ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE)
 #define IS_INTEGER_TYPE(_t)          ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)))
 
 #define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
-#define IS_MATHABLE_TYPE(_t) (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
+#define IS_MATHABLE_TYPE(_t) \
+  (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
 
 #define IS_VALID_TINYINT(_t)   ((_t) >= INT8_MIN && (_t) <= INT8_MAX)
 #define IS_VALID_SMALLINT(_t)  ((_t) >= INT16_MIN && (_t) <= INT16_MAX)
@@ -244,7 +249,7 @@ typedef struct tDataTypeDescriptor {
   int16_t type;
   int16_t nameLen;
   int32_t bytes;
-  char   *name;
+  char *  name;
   int64_t minValue;
   int64_t maxValue;
   int32_t (*compFunc)(const char *const input, int32_t inputSize, const int32_t nelements, char *const output,
@@ -277,4 +282,4 @@ void *getDataMax(int32_t type);
 }
 #endif
 
-#endif  /*_TD_COMMON_TTYPE_H_*/
+#endif /*_TD_COMMON_TTYPE_H_*/

source/client/src/clientImpl.c

@@ -923,7 +923,7 @@ void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
     int32_t bytes = pResultInfo->fields[i].bytes;
 
     if (IS_VAR_DATA_TYPE(type)) {
-      if (pCol->offset[pResultInfo->current] != -1) {
+      if (!IS_VAR_NULL_TYPE(type, bytes) && pCol->offset[pResultInfo->current] != -1) {
         char* pStart = pResultInfo->pCol[i].offset[pResultInfo->current] + pResultInfo->pCol[i].pData;
 
         pResultInfo->length[i] = varDataLen(pStart);

source/common/src/tmsgcb.c

@@ -22,41 +22,21 @@ static SMsgCb defaultMsgCb;
 void tmsgSetDefault(const SMsgCb* msgcb) { defaultMsgCb = *msgcb; }
 
 int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg) {
-  PutToQueueFp fp = msgcb->queueFps[qtype];
-  return (*fp)(msgcb->mgmt, pMsg);
+  return (*msgcb->putToQueueFp)(msgcb->mgmt, qtype, pMsg);
 }
 
 int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype) {
-  GetQueueSizeFp fp = msgcb->qsizeFp;
-  return (*fp)(msgcb->mgmt, vgId, qtype);
+  return (*msgcb->qsizeFp)(msgcb->mgmt, vgId, qtype);
 }
 
-int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) {
-  SendReqFp fp = defaultMsgCb.sendReqFp;
-  return (*fp)(epSet, pMsg);
-}
+int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) { return (*defaultMsgCb.sendReqFp)(epSet, pMsg); }
 
-void tmsgSendRsp(SRpcMsg* pMsg) {
-  SendRspFp fp = defaultMsgCb.sendRspFp;
-  return (*fp)(pMsg);
-}
+void tmsgSendRsp(SRpcMsg* pMsg) { return (*defaultMsgCb.sendRspFp)(pMsg); }
 
-void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) {
-  SendRedirectRspFp fp = defaultMsgCb.sendRedirectRspFp;
-  (*fp)(pMsg, pNewEpSet);
-}
+void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { (*defaultMsgCb.sendRedirectRspFp)(pMsg, pNewEpSet); }
 
-void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) {
-  RegisterBrokenLinkArgFp fp = defaultMsgCb.registerBrokenLinkArgFp;
-  (*fp)(pMsg);
-}
+void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { (*defaultMsgCb.registerBrokenLinkArgFp)(pMsg); }
 
-void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) {
-  ReleaseHandleFp fp = defaultMsgCb.releaseHandleFp;
-  (*fp)(pHandle, type);
-}
+void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { (*defaultMsgCb.releaseHandleFp)(pHandle, type); }
 
-void tmsgReportStartup(const char* name, const char* desc) {
-  ReportStartup fp = defaultMsgCb.reportStartupFp;
+void tmsgReportStartup(const char* name, const char* desc) { (*defaultMsgCb.reportStartupFp)(name, desc); }
\ No newline at end of file
-  (*fp)(name, desc);
-}
\ No newline at end of file

source/dnode/mgmt/mgmt_mnode/inc/mmInt.h

@@ -65,10 +65,7 @@ int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc);
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -134,7 +134,7 @@ SArray *mmGetMsgHandles() {
 
   // Requests handled by DNODE
   if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_DND_ALTER_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_MNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_QNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_QNODE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
@@ -190,7 +190,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_TRANS, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH, mmPutNodeMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_DND_ALTER_MNODE, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_TOPIC, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_mnode/src/mmInt.c

@@ -105,10 +105,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->path = pInput->path;
   pMgmt->name = pInput->name;
   pMgmt->msgCb = pInput->msgCb;
-  pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)mmPutRpcMsgToQueryQueue;
-  pMgmt->msgCb.queueFps[READ_QUEUE] = (PutToQueueFp)mmPutRpcMsgToReadQueue;
-  pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue;
-  pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToSyncQueue;
+  pMgmt->msgCb.putToQueueFp = (PutToQueueFp)mmPutRpcMsgToQueue;
   pMgmt->msgCb.mgmt = pMgmt;
   taosThreadRwlockInit(&pMgmt->lock, NULL);
 

source/dnode/mgmt/mgmt_mnode/src/mmWorker.c

@@ -96,40 +96,38 @@ int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   return mmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg);
 }
 
-static inline int32_t mmPutRpcMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pRpc) {
+int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
   if (pMsg == NULL) return -1;
-
-  dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType));
   memcpy(pMsg, pRpc, sizeof(SRpcMsg));
-  taosWriteQitem(pWorker->queue, pMsg);
-  return 0;
-}
 
-int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutRpcMsgToWorker(&pMgmt->queryWorker, pMsg);
-}
-
-int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutRpcMsgToWorker(&pMgmt->writeWorker, pMsg);
-}
-
-int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return mmPutRpcMsgToWorker(&pMgmt->readWorker, pMsg);
-}
-
-int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  int32_t code = -1;
-  if (mmAcquire(pMgmt) == 0) {
-    code = mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg);
-    mmRelease(pMgmt);
-  }
-
-  if (code != 0) {
-    rpcFreeCont(pMsg->pCont);
-    pMsg->pCont = NULL;
+  switch (qtype) {
+    case WRITE_QUEUE:
+      dTrace("msg:%p, is created and will put into vnode-write queue", pMsg);
+      taosWriteQitem(pMgmt->writeWorker.queue, pMsg);
+      return 0;
+    case QUERY_QUEUE:
+      dTrace("msg:%p, is created and will put into vnode-query queue", pMsg);
+      taosWriteQitem(pMgmt->queryWorker.queue, pMsg);
+      return 0;
+
+    case READ_QUEUE:
+      dTrace("msg:%p, is created and will put into vnode-read queue", pMsg);
+      taosWriteQitem(pMgmt->readWorker.queue, pMsg);
+      return 0;
+    case SYNC_QUEUE:
+      if (mmAcquire(pMgmt) == 0) {
+        dTrace("msg:%p, is created and will put into vnode-sync queue", pMsg);
+        taosWriteQitem(pMgmt->syncWorker.queue, pMsg);
+        mmRelease(pMgmt);
+        return 0;
+      } else {
+        return -1;
+      }
+    default:
+      terrno = TSDB_CODE_INVALID_PARA;
+      return -1;
   }
-  return code;
 }
 
 int32_t mmStartWorker(SMnodeMgmt *pMgmt) {

source/dnode/mgmt/mgmt_qnode/inc/qmInt.h

@@ -42,8 +42,7 @@ int32_t qmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // qmWorker.c
-int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pMsg);
 int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype);
 
 int32_t qmStartWorker(SQnodeMgmt *pMgmt);

source/dnode/mgmt/mgmt_qnode/src/qmInt.c

@@ -43,8 +43,7 @@ static int32_t qmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->path = pInput->path;
   pMgmt->name = pInput->name;
   pMgmt->msgCb = pInput->msgCb;
-  pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qmPutRpcMsgToQueryQueue;
-  pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)qmPutRpcMsgToFetchQueue;
+  pMgmt->msgCb.putToQueueFp = (PutToQueueFp)qmPutRpcMsgToQueue;
   pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)qmGetQueueSize;
   pMgmt->msgCb.mgmt = pMgmt;
 

source/dnode/mgmt/mgmt_qnode/src/qmWorker.c

@@ -68,22 +68,24 @@ int32_t qmPutNodeMsgToMonitorQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   return qmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg);
 }
 
-static int32_t qmPutRpcMsgToWorker(SQnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pRpc) {
+int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
   if (pMsg == NULL) return -1;
-
-  dTrace("msg:%p, create and put into worker:%s, type:%s", pMsg, pWorker->name, TMSG_INFO(pRpc->msgType));
   memcpy(pMsg, pRpc, sizeof(SRpcMsg));
-  taosWriteQitem(pWorker->queue, pMsg);
-  return 0;
-}
-
-int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return qmPutRpcMsgToWorker(pMgmt, &pMgmt->queryWorker, pRpc);
-}
 
-int32_t qmPutRpcMsgToFetchQueue(SQnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return qmPutRpcMsgToWorker(pMgmt, &pMgmt->fetchWorker, pRpc);
+  switch (qtype) {
+    case QUERY_QUEUE:
+      dTrace("msg:%p, is created and will put into qnode-query queue", pMsg);
+      taosWriteQitem(pMgmt->queryWorker.queue, pMsg);
+      return 0;
+    case READ_QUEUE:
+      dTrace("msg:%p, is created and will put into qnode-fetch queue", pMsg);
+      taosWriteQitem(pMgmt->fetchWorker.queue, pMsg);
+      return 0;
+    default:
+      terrno = TSDB_CODE_INVALID_PARA;
+      return -1;
+  }
 }
 
 int32_t qmGetQueueSize(SQnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {

source/dnode/mgmt/mgmt_vnode/inc/vmInt.h

@@ -44,26 +44,26 @@ typedef struct SVnodeMgmt {
 } SVnodeMgmt;
 
 typedef struct {
-  int32_t  vgId;
-  int32_t  vgVersion;
-  int8_t   dropped;
-  char     path[PATH_MAX + 20];
+  int32_t vgId;
+  int32_t vgVersion;
+  int8_t  dropped;
+  char    path[PATH_MAX + 20];
 } SWrapperCfg;
 
 typedef struct {
-  int32_t       vgId;
-  int32_t       refCount;
-  int32_t       vgVersion;
-  int8_t        dropped;
-  int8_t        accessState;
-  char         *path;
-  SVnode       *pImpl;
-  STaosQueue   *pWriteQ;
-  STaosQueue   *pSyncQ;
-  STaosQueue   *pApplyQ;
-  STaosQueue   *pQueryQ;
-  STaosQueue   *pFetchQ;
-  STaosQueue   *pMergeQ;
+  int32_t     vgId;
+  int32_t     refCount;
+  int32_t     vgVersion;
+  int8_t      dropped;
+  int8_t      accessState;
+  char       *path;
+  SVnode     *pImpl;
+  STaosQueue *pWriteQ;
+  STaosQueue *pSyncQ;
+  STaosQueue *pApplyQ;
+  STaosQueue *pQueryQ;
+  STaosQueue *pFetchQ;
+  STaosQueue *pMergeQ;
 } SVnodeObj;
 
 typedef struct {
@@ -100,21 +100,16 @@ void    vmStopWorker(SVnodeMgmt *pMgmt);
 int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
 void    vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
 
-int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype);
-
-int32_t vmPutNodeMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc);
+
+int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 #ifdef __cplusplus
 }

source/dnode/mgmt/mgmt_vnode/src/vmFile.c

@@ -16,6 +16,8 @@
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
 
+#define MAX_CONTENT_LEN 1024 * 1024
+
 SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
   taosThreadRwlockRdlock(&pMgmt->lock);
 
@@ -29,7 +31,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
     SVnodeObj  *pVnode = *ppVnode;
     if (pVnode && num < size) {
       int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
-      dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
+      // dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
       pVnodes[num] = (*ppVnode);
       num++;
       pIter = taosHashIterate(pMgmt->hash, pIter);
@@ -47,7 +49,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
 int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
   int32_t      code = TSDB_CODE_INVALID_JSON_FORMAT;
   int32_t      len = 0;
-  int32_t      maxLen = 1024 * 1024;
+  int32_t      maxLen = MAX_CONTENT_LEN;
   char        *content = taosMemoryCalloc(1, maxLen + 1);
   cJSON       *root = NULL;
   FILE        *fp = NULL;
@@ -128,7 +130,7 @@ int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t
 
   *numOfVnodes = vnodesNum;
   code = 0;
-  dDebug("succcessed to read file %s", file);
+  dDebug("succcessed to read file %s, numOfVnodes:%d", file, vnodesNum);
 
 _OVER:
   if (content != NULL) taosMemoryFree(content);
@@ -156,7 +158,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
   SVnodeObj **pVnodes = vmGetVnodeListFromHash(pMgmt, &numOfVnodes);
 
   int32_t len = 0;
-  int32_t maxLen = 1024 * 1024;
+  int32_t maxLen = MAX_CONTENT_LEN;
   char   *content = taosMemoryCalloc(1, maxLen + 1);
   if (content == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -195,6 +197,6 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
     taosMemoryFree(pVnodes);
   }
 
-  dDebug("successed to write %s", realfile);
+  dDebug("successed to write %s, numOfVnodes:%d", realfile, numOfVnodes);
   return taosRenameFile(file, realfile);
 }
\ No newline at end of file

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -210,7 +210,7 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
-  dDebug("vgId:%d, create vnode req is received, tsma:%d standby:%d", createReq.vgId, createReq.isTsma,
+  dDebug("vgId:%d, start to create vnode, tsma:%d standby:%d", createReq.vgId, createReq.isTsma,
          createReq.standby);
   vmGenerateVnodeCfg(&createReq, &vnodeCfg);
 
@@ -294,7 +294,7 @@ int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   int32_t vgId = dropReq.vgId;
-  dDebug("vgId:%d, drop vnode req is received", vgId);
+  dDebug("vgId:%d, start to drop vnode", vgId);
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
   if (pVnode == NULL) {
@@ -321,59 +321,59 @@ SArray *vmGetMsgHandles() {
   SArray *pArray = taosArrayInit(32, sizeof(SMgmtHandle));
   if (pArray == NULL) goto _OVER;
 
-  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_INFO, vmPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_LOAD, vmPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_INFO, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MON_VM_LOAD, vmPutMsgToMonitorQueue, 0) == NULL) goto _OVER;
 
   // Requests handled by VNODE
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONSUME, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_QUERY, vmPutNodeMsgToQueryQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONNECT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_DISCONNECT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  // if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_SET_CUR, vmPutNodeMsgToWriteQueue, 0)== NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RUN, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER;
-
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
-
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES_REPLY, vmPutNodeMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_CONTINUE, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONSUME, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_CONNECT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_DISCONNECT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  // if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_SET_CUR, vmPutMsgToWriteQueue, 0)== NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RUN, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DISPATCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_RECOVER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
+
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_TIMEOUT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_PING_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_REQUEST_VOTE_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
 
   code = 0;
 

source/dnode/mgmt/mgmt_vnode/src/vmInt.c

@@ -18,21 +18,17 @@
 
 SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
-  int32_t    refCount = 0;
 
   taosThreadRwlockRdlock(&pMgmt->lock);
   taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
   if (pVnode == NULL) {
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
   } else {
-    refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
+    int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
+    // dTrace("vgId:%d, acquire vnode, ref:%d", pVnode->vgId, refCount);
   }
   taosThreadRwlockUnlock(&pMgmt->lock);
 
-  if (pVnode != NULL) {
-    dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
-  }
-
   return pVnode;
 }
 
@@ -41,8 +37,8 @@ void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
 
   taosThreadRwlockRdlock(&pMgmt->lock);
   int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
+  // dTrace("vgId:%d, release vnode, ref:%d", pVnode->vgId, refCount);
   taosThreadRwlockUnlock(&pMgmt->lock);
-  dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
 }
 
 int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
@@ -138,7 +134,7 @@ static void *vmOpenVnodeInThread(void *param) {
     }
   }
 
-  dDebug("thread:%d, total vnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened,
+  dDebug("thread:%d, numOfVnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened,
          pThread->failed);
   return NULL;
 }
@@ -160,7 +156,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
 
   pMgmt->state.totalVnodes = numOfVnodes;
 
-  int32_t threadNum = 1;  // tsNumOfCores;
+  int32_t threadNum = 1;
   int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
 
   SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread));
@@ -253,12 +249,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->path = pInput->path;
   pMgmt->name = pInput->name;
   pMgmt->msgCb = pInput->msgCb;
-  pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToWriteQueue;
-  pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)vmPutRpcMsgToSyncQueue;
-  pMgmt->msgCb.queueFps[APPLY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToApplyQueue;
-  pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)vmPutRpcMsgToQueryQueue;
-  pMgmt->msgCb.queueFps[FETCH_QUEUE] = (PutToQueueFp)vmPutRpcMsgToFetchQueue;
-  pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue;
+  pMgmt->msgCb.putToQueueFp = (PutToQueueFp)vmPutRpcMsgToQueue;
   pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize;
   pMgmt->msgCb.mgmt = pMgmt;
   taosThreadRwlockInit(&pMgmt->lock, NULL);

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -29,11 +29,11 @@ static inline void vmSendRsp(SRpcMsg *pMsg, int32_t code) {
   tmsgSendRsp(&rsp);
 }
 
-static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
+static void vmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SVnodeMgmt *pMgmt = pInfo->ahandle;
   int32_t     code = -1;
-  dTrace("msg:%p, get from vnode queue, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
 
+  dTrace("msg:%p, get from vnode-mgmt queue", pMsg);
   switch (pMsg->msgType) {
     case TDMT_MON_VM_INFO:
       code = vmProcessGetMonitorInfoReq(pMgmt, pMsg);
@@ -49,11 +49,14 @@ static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
       break;
     default:
       terrno = TSDB_CODE_MSG_NOT_PROCESSED;
-      dError("msg:%p, not processed in vnode queue", pMsg);
+      dError("msg:%p, not processed in vnode-mgmt queue", pMsg);
   }
 
   if (IsReq(pMsg)) {
-    if (code != 0 && terrno != 0) code = terrno;
+    if (code != 0 && terrno != 0) {
+      dError("msg:%p failed to process since %s", pMsg, terrstr());
+      code = terrno;
+    }
     vmSendRsp(pMsg, code);
   }
 
@@ -65,13 +68,15 @@ static void vmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
 static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SVnodeObj *pVnode = pInfo->ahandle;
 
-  dTrace("msg:%p, get from vnode-query queue", pMsg);
+  dTrace("vgId:%d, msg:%p get from vnode-query queue", pVnode->vgId, pMsg);
   int32_t code = vnodeProcessQueryMsg(pVnode->pImpl, pMsg);
   if (code != 0) {
     if (terrno != 0) code = terrno;
+    dError("vgId:%d, msg:%p failed to query since %s", pVnode->vgId, pMsg, terrstr());
     vmSendRsp(pMsg, code);
   }
-  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
+
+  dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
   rpcFreeCont(pMsg->pCont);
   taosFreeQitem(pMsg);
 }
@@ -79,13 +84,15 @@ static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
 static void vmProcessFetchQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   SVnodeObj *pVnode = pInfo->ahandle;
 
-  dTrace("msg:%p, get from vnode-fetch queue", pMsg);
+  dTrace("vgId:%d, msg:%p get from vnode-fetch queue", pVnode->vgId, pMsg);
   int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
   if (code != 0) {
     if (terrno != 0) code = terrno;
+    dError("vgId:%d, msg:%p failed to fetch since %s", pVnode->vgId, pMsg, terrstr());
     vmSendRsp(pMsg, code);
   }
-  dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
+
+  dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
   rpcFreeCont(pMsg->pCont);
   taosFreeQitem(pMsg);
 }
@@ -99,7 +106,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
   for (int32_t m = 0; m < numOfMsgs; m++) {
     if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
-    dTrace("vgId:%d, get msg:%p from vnode-write queue, type:%s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType));
+    dTrace("vgId:%d, msg:%p get from vnode-write queue", pVnode->vgId, pMsg);
 
     if (taosArrayPush(pArray, &pMsg) == NULL) {
       dError("vgId:%d, failed to push msg:%p to vnode-write array", pVnode->vgId, pMsg);
@@ -113,7 +120,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
     if (code == TSDB_CODE_ACTION_IN_PROGRESS) continue;
     if (code != 0) {
-      dError("vgId:%d, msg:%p failed to process since %s", pVnode->vgId, pMsg, tstrerror(code));
+      dError("vgId:%d, msg:%p failed to write since %s", pVnode->vgId, pMsg, tstrerror(code));
       vmSendRsp(pMsg, code);
       continue;
     }
@@ -129,14 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
       SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info};
       tmsgSendRedirectRsp(&rsp, &newEpSet);
     } else {
-      dError("vgId:%d, msg:%p failed to process since %s", pVnode->vgId, pMsg, tstrerror(code));
+      dError("vgId:%d, msg:%p failed to write since %s", pVnode->vgId, pMsg, tstrerror(code));
       vmSendRsp(pMsg, code);
     }
   }
 
   for (int32_t i = 0; i < numOfMsgs; i++) {
     pMsg = *(SRpcMsg **)taosArrayGet(pArray, i);
-    dTrace("vgId:%d, msg:%p, is freed", pVnode->vgId, pMsg);
+    dTrace("vgId:%d, msg:%p is freed", pVnode->vgId, pMsg);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
   }
@@ -146,10 +153,11 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
+  SRpcMsg   *pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
-    SRpcMsg *pMsg = NULL;
-    taosGetQitem(qall, (void **)&pMsg);
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+    dTrace("vgId:%d, msg:%p get from vnode-apply queue", pVnode->vgId, pMsg);
 
     // init response rpc msg
     SRpcMsg rsp = {0};
@@ -164,7 +172,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     // apply data into tsdb
     if (vnodeProcessWriteReq(pVnode->pImpl, &originalRpcMsg, pSyncApplyMsg->fsmMeta.index, &rsp) < 0) {
       rsp.code = terrno;
-      dTrace("msg:%p, process write error since %s", pMsg, terrstr());
+      dError("vgId:%d, msg:%p failed to apply since %s", pVnode->vgId, pMsg, terrstr());
     }
 
     syncApplyMsgDestroy(pSyncApplyMsg);
@@ -176,6 +184,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
       tmsgSendRsp(&rsp);
     }
 
+    dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, rsp.code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
   }
@@ -183,23 +192,22 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
+  SRpcMsg   *pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
-    SRpcMsg *pMsg = NULL;
-    taosGetQitem(qall, (void **)&pMsg);
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+    dTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg);
 
     int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL);
     if (code != 0) {
+      dError("vgId:%d, msg:%p failed to sync since %s", pVnode->vgId, pMsg, terrstr());
       if (pMsg->info.handle != NULL) {
-        SRpcMsg rsp = {
-            .code = (terrno < 0) ? terrno : code,
-            .info = pMsg->info,
-        };
-        dTrace("msg:%p, failed to process sync queue since %s", pMsg, terrstr());
-        tmsgSendRsp(&rsp);
+        if (terrno != 0) code = terrno;
+        vmSendRsp(pMsg, code);
       }
     }
 
+    dTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
   }
@@ -207,24 +215,26 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
 
 static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
+  SRpcMsg   *pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
-    SRpcMsg *pMsg = NULL;
-    taosGetQitem(qall, (void **)&pMsg);
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+    dTrace("vgId:%d, msg:%p get from vnode-merge queue", pVnode->vgId, pMsg);
 
-    dTrace("msg:%p, get from vnode-merge queue", pMsg);
     int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
     if (code != 0) {
+      dError("vgId:%d, msg:%p failed to merge since %s", pVnode->vgId, pMsg, terrstr());
       if (terrno != 0) code = terrno;
       vmSendRsp(pMsg, code);
     }
+
     dTrace("msg:%p, is freed, code:0x%x", pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);
   }
 }
 
-static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {
+static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) {
   SMsgHead *pHead = pMsg->pCont;
   int32_t   code = 0;
 
@@ -233,31 +243,36 @@ static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) {
-    dError("vgId:%d, failed to put msg:%p into vnode-queue since %s", pHead->vgId, pMsg, terrstr());
+    dError("vgId:%d, failed to put msg:%p into vnode queue since %s, type:%s", pHead->vgId, pMsg, terrstr(),
+           TMSG_INFO(pMsg->msgType));
     return terrno != 0 ? terrno : -1;
   }
 
   switch (qtype) {
     case QUERY_QUEUE:
-      dTrace("msg:%p, put into vnode-query worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      dTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pQueryQ, pMsg);
       break;
     case FETCH_QUEUE:
-      dTrace("msg:%p, put into vnode-fetch worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      dTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pFetchQ, pMsg);
       break;
     case WRITE_QUEUE:
-      dTrace("msg:%p, put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      dTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pWriteQ, pMsg);
       break;
     case SYNC_QUEUE:
-      dTrace("msg:%p, put into vnode-sync worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      dTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pSyncQ, pMsg);
       break;
     case MERGE_QUEUE:
-      dTrace("msg:%p, put into vnode-merge worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      dTrace("vgId:%d, msg:%p put into vnode-merge queue", pVnode->vgId, pMsg);
       taosWriteQitem(pVnode->pMergeQ, pMsg);
       break;
+    case APPLY_QUEUE:
+      dTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg);
+      taosWriteQitem(pVnode->pApplyQ, pMsg);
+      break;
     default:
       code = -1;
       terrno = TSDB_CODE_INVALID_PARA;
@@ -268,110 +283,39 @@ static int32_t vmPutNodeMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType
   return code;
 }
 
-int32_t vmPutNodeMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutNodeMsgToQueue(pMgmt, pMsg, SYNC_QUEUE);
-}
+int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, SYNC_QUEUE); }
 
-int32_t vmPutNodeMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutNodeMsgToQueue(pMgmt, pMsg, WRITE_QUEUE);
-}
+int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, WRITE_QUEUE); }
 
-int32_t vmPutNodeMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutNodeMsgToQueue(pMgmt, pMsg, QUERY_QUEUE);
-}
+int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, QUERY_QUEUE); }
 
-int32_t vmPutNodeMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutNodeMsgToQueue(pMgmt, pMsg, FETCH_QUEUE);
-}
+int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }
 
-int32_t vmPutNodeMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  return vmPutNodeMsgToQueue(pMgmt, pMsg, MERGE_QUEUE);
-}
+int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); }
 
-int32_t vmPutNodeMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SSingleWorker *pWorker = &pMgmt->mgmtWorker;
-  dTrace("msg:%p, put into vnode-mgmt worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-  taosWriteQitem(pWorker->queue, pMsg);
+int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  dTrace("msg:%p, put into vnode-mgmt queue", pMsg);
+  taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg);
   return 0;
 }
 
-int32_t vmPutNodeMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SSingleWorker *pWorker = &pMgmt->monitorWorker;
-  dTrace("msg:%p, put into vnode-monitor worker, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
-  taosWriteQitem(pWorker->queue, pMsg);
+int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
+  dTrace("msg:%p, put into vnode-monitor queue", pMsg);
+  taosWriteQitem(pMgmt->monitorWorker.queue, pMsg);
   return 0;
 }
 
-static int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc, EQueueType qtype) {
-  SMsgHead  *pHead = pRpc->pCont;
-  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
-  if (pVnode == NULL) return -1;
-
+int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
-  int32_t  code = 0;
-
-  if (pMsg == NULL) {
-    rpcFreeCont(pRpc->pCont);
-    pRpc->pCont = NULL;
-    code = -1;
-  } else {
-    memcpy(pMsg, pRpc, sizeof(SRpcMsg));
-    switch (qtype) {
-      case WRITE_QUEUE:
-        dTrace("msg:%p, create and put into vnode-write worker, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pWriteQ, pMsg);
-        break;
-      case QUERY_QUEUE:
-        dTrace("msg:%p, create and put into vnode-query queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pQueryQ, pMsg);
-        break;
-      case FETCH_QUEUE:
-        dTrace("msg:%p, create and put into vnode-fetch queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pFetchQ, pMsg);
-        break;
-      case APPLY_QUEUE:
-        dTrace("msg:%p, create and put into vnode-apply queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pApplyQ, pMsg);
-        break;
-      case MERGE_QUEUE:
-        dTrace("msg:%p, create and put into vnode-merge queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pMergeQ, pMsg);
-        break;
-      case SYNC_QUEUE:
-        dTrace("msg:%p, create and put into vnode-sync queue, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
-        taosWriteQitem(pVnode->pSyncQ, pMsg);
-        break;
-      default:
-        code = -1;
-        terrno = TSDB_CODE_INVALID_PARA;
-        break;
-    }
-  }
-
-  vmReleaseVnode(pMgmt, pVnode);
-  return code;
-}
-
-int32_t vmPutRpcMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return vmPutRpcMsgToQueue(pMgmt, pRpc, WRITE_QUEUE);
-}
-
-int32_t vmPutRpcMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) { return vmPutRpcMsgToQueue(pMgmt, pRpc, SYNC_QUEUE); }
+  if (pMsg == NULL) return -1;
 
-int32_t vmPutRpcMsgToApplyQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return vmPutRpcMsgToQueue(pMgmt, pRpc, APPLY_QUEUE);
-}
-
-int32_t vmPutRpcMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return vmPutRpcMsgToQueue(pMgmt, pRpc, QUERY_QUEUE);
-}
+  SMsgHead *pHead = pRpc->pCont;
+  dTrace("vgId:%d, msg:%p is created, type:%s", pHead->vgId, pMsg, TMSG_INFO(pMsg->msgType));
 
-int32_t vmPutRpcMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return vmPutRpcMsgToQueue(pMgmt, pRpc, FETCH_QUEUE);
-}
-
-int32_t vmPutRpcMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pRpc) {
-  return vmPutRpcMsgToQueue(pMgmt, pRpc, MERGE_QUEUE);
+  pHead->contLen = htonl(pHead->contLen);
+  pHead->vgId = htonl(pHead->vgId);
+  memcpy(pMsg, pRpc, sizeof(SRpcMsg));
+  return vmPutMsgToQueue(pMgmt, pMsg, qtype);
 }
 
 int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
@@ -467,29 +411,23 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
   pMPool->max = tsNumOfVnodeMergeThreads;
   if (tWWorkerInit(pMPool) != 0) return -1;
 
-  SSingleWorkerCfg cfg = {
+  SSingleWorkerCfg mgmtCfg = {
       .min = 1,
       .max = 1,
       .name = "vnode-mgmt",
-      .fp = (FItem)vmProcessQueue,
+      .fp = (FItem)vmProcessMgmtQueue,
       .param = pMgmt,
   };
-  if (tSingleWorkerInit(&pMgmt->mgmtWorker, &cfg) != 0) {
-    dError("failed to start vnode-mgmt worker since %s", terrstr());
-    return -1;
-  }
+  if (tSingleWorkerInit(&pMgmt->mgmtWorker, &mgmtCfg) != 0) return -1;
 
-  SSingleWorkerCfg mCfg = {
+  SSingleWorkerCfg monitorCfg = {
       .min = 1,
       .max = 1,
       .name = "vnode-monitor",
-      .fp = (FItem)vmProcessQueue,
+      .fp = (FItem)vmProcessMgmtQueue,
       .param = pMgmt,
   };
-  if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
-    dError("failed to start vnode-monitor worker since %s", terrstr());
-    return -1;
-  }
+  if (tSingleWorkerInit(&pMgmt->monitorWorker, &monitorCfg) != 0) return -1;
 
   dDebug("vnode workers are initialized");
   return 0;

source/dnode/mgmt/node_mgmt/src/dmMgmt.c

@@ -229,7 +229,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
   taosThreadRwlockRdlock(&pWrapper->lock);
   if (pWrapper->deployed) {
     int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
-    dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount);
+    // dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount);
   } else {
     terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
     pRetWrapper = NULL;
@@ -245,7 +245,7 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
   taosThreadRwlockRdlock(&pWrapper->lock);
   if (pWrapper->deployed || (InParentProc(pWrapper) && pWrapper->required)) {
     int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
-    dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount);
+    // dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount);
   } else {
     terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
     code = -1;
@@ -261,7 +261,7 @@ void dmReleaseWrapper(SMgmtWrapper *pWrapper) {
   taosThreadRwlockRdlock(&pWrapper->lock);
   int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);
   taosThreadRwlockUnlock(&pWrapper->lock);
-  dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
+  // dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
 }
 
 static void dmGetServerStartupStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -117,6 +117,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
   if (pMsg == NULL) {
     goto _OVER;
   }
+  dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
 
   if (dmBuildNodeMsg(pMsg, pRpc) != 0) {
     goto _OVER;

source/dnode/mgmt/test/mnode/dmnode.cpp

@@ -94,7 +94,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
     void*   pReq = rpcMallocCont(contLen);
     tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
 
-    SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen);
     ASSERT_NE(pRsp, nullptr);
     ASSERT_EQ(pRsp->code, TSDB_CODE_INVALID_OPTION);
   }
@@ -111,7 +111,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
     void*   pReq = rpcMallocCont(contLen);
     tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
 
-    SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen);
     ASSERT_NE(pRsp, nullptr);
     ASSERT_EQ(pRsp->code, TSDB_CODE_INVALID_OPTION);
   }
@@ -128,7 +128,7 @@ TEST_F(DndTestMnode, 02_Alter_Mnode) {
     void*   pReq = rpcMallocCont(contLen);
     tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
 
-    SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen);
     ASSERT_NE(pRsp, nullptr);
     ASSERT_EQ(pRsp->code, 0);
   }
@@ -186,7 +186,7 @@ TEST_F(DndTestMnode, 03_Drop_Mnode) {
     void*   pReq = rpcMallocCont(contLen);
     tSerializeSDCreateMnodeReq(pReq, contLen, &alterReq);
 
-    SRpcMsg* pRsp = test.SendReq(TDMT_DND_ALTER_MNODE, pReq, contLen);
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_MNODE, pReq, contLen);
     ASSERT_NE(pRsp, nullptr);
     ASSERT_EQ(pRsp->code, TSDB_CODE_NODE_NOT_DEPLOYED);
   }

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

@@ -53,10 +53,10 @@ int32_t mndInitMnode(SMnode *pMnode) {
   };
 
   mndSetMsgHandle(pMnode, TDMT_MND_CREATE_MNODE, mndProcessCreateMnodeReq);
-  mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE, mndProcessAlterMnodeReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE, mndProcessAlterMnodeReq);
   mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq);
   mndSetMsgHandle(pMnode, TDMT_DND_CREATE_MNODE_RSP, mndProcessCreateMnodeRsp);
-  mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE_RSP, mndProcessAlterMnodeRsp);
+  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndProcessAlterMnodeRsp);
   mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndProcessDropMnodeRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes);
@@ -338,7 +338,7 @@ static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDno
         .epSet = alterEpset,
         .pCont = pReq,
         .contLen = contLen,
-        .msgType = TDMT_DND_ALTER_MNODE,
+        .msgType = TDMT_MND_ALTER_MNODE,
         .acceptableCode = 0,
     };
 
@@ -506,7 +506,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
         .epSet = alterEpset,
         .pCont = pReq,
         .contLen = contLen,
-        .msgType = TDMT_DND_ALTER_MNODE,
+        .msgType = TDMT_MND_ALTER_MNODE,
         .acceptableCode = 0,
     };
 

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

@@ -576,15 +576,15 @@ SArray *metaGetSmaTbUids(SMeta *pMeta) {
 const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) {
   ASSERT(pEntry->type == TSDB_CHILD_TABLE);
   STag *tag = (STag *)pEntry->ctbEntry.pTags;
-  if (type == TSDB_DATA_TYPE_JSON){
-    if(tag->nTag == 0){
+  if (type == TSDB_DATA_TYPE_JSON) {
+    if (tag->nTag == 0) {
       return NULL;
     }
     return tag;
   }
   bool find = tTagGet(tag, val);
 
-  if(!find){
+  if (!find) {
     return NULL;
   }
   return val;
@@ -605,8 +605,6 @@ typedef struct {
 int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
   SIdxCursor *pCursor = NULL;
 
-  char *tagData = param->val;
-
   int32_t ret = 0, valid = 0;
   pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
   pCursor->pMeta = pMeta;
@@ -623,12 +621,16 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
   int32_t     nKey = 0;
 
   int32_t nTagData = 0;
-  if(IS_VAR_DATA_TYPE(param->type)){
-    nTagData = strlen(param->val);
-  }else{
+  void *  tagData = NULL;
+
+  if (IS_VAR_DATA_TYPE(param->type)) {
+    tagData = varDataVal(param->val);
+    nTagData = varDataLen(param->val);
+  } else {
+    tagData = param->val;
     nTagData = tDataTypes[param->type].bytes;
   }
-  ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, param->val, nTagData, pCursor->type,
+  ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, tagData, nTagData, pCursor->type,
                             param->reverse ? INT64_MAX : INT64_MIN, &pKey, &nKey);
   if (ret != 0) {
     goto END;
@@ -637,6 +639,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
   if (tdbTbcMoveTo(pCursor->pCur, pKey, nKey, &cmp) < 0) {
     goto END;
   }
+
   void *  entryKey = NULL, *entryVal = NULL;
   int32_t nEntryKey, nEntryVal;
   while (1) {
@@ -649,7 +652,12 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
       int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);
       if (cmp == 0) {
         // match
-        tb_uid_t tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
+        tb_uid_t tuid = 0;
+        if (IS_VAR_DATA_TYPE(pKey->type)) {
+          tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data));
+        } else {
+          tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
+        }
         taosArrayPush(pUids, &tuid);
       } else if (cmp == 1) {
         // not match but should continue to iter
@@ -670,4 +678,4 @@ END:
   taosMemoryFree(pCursor);
 
   return ret;
-}
+}
\ No newline at end of file

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

@@ -49,9 +49,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   int         vLen = 0;
   const void *pKey = NULL;
   const void *pVal = NULL;
-  void       *pBuf = NULL;
+  void *      pBuf = NULL;
   int32_t     szBuf = 0;
-  void       *p = NULL;
+  void *      p = NULL;
   SMetaReader mr = {0};
 
   // validate req
@@ -105,7 +105,7 @@ int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
   }
 
   // drop all child tables
-  TBC    *pCtbIdxc = NULL;
+  TBC *   pCtbIdxc = NULL;
   SArray *pArray = taosArrayInit(8, sizeof(tb_uid_t));
 
   tdbTbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
@@ -160,8 +160,8 @@ _exit:
 int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   SMetaEntry  oStbEntry = {0};
   SMetaEntry  nStbEntry = {0};
-  TBC        *pUidIdxc = NULL;
-  TBC        *pTbDbc = NULL;
+  TBC *       pUidIdxc = NULL;
+  TBC *       pTbDbc = NULL;
   const void *pData;
   int         nData;
   int64_t     oversion;
@@ -280,7 +280,7 @@ _err:
 }
 
 int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
-  void    *pData = NULL;
+  void *   pData = NULL;
   int      nData = 0;
   int      rc = 0;
   tb_uid_t uid;
@@ -306,7 +306,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
 }
 
 static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
-  void      *pData = NULL;
+  void *     pData = NULL;
   int        nData = 0;
   int        rc = 0;
   int64_t    version;
@@ -341,15 +341,16 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
   return 0;
 }
 
+
 static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq, STableMetaRsp *pMetaRsp) {
   void *          pVal = NULL;
   int             nVal = 0;
-  const void     *pData = NULL;
+  const void *    pData = NULL;
   int             nData = 0;
   int             ret = 0;
   tb_uid_t        uid;
   int64_t         oversion;
-  SSchema        *pColumn = NULL;
+  SSchema *       pColumn = NULL;
   SMetaEntry      entry = {0};
   SSchemaWrapper *pSchema;
   int             c;
@@ -499,7 +500,7 @@ _err:
 static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
   SMetaEntry  ctbEntry = {0};
   SMetaEntry  stbEntry = {0};
-  void       *pVal = NULL;
+  void *      pVal = NULL;
   int         nVal = 0;
   int         ret;
   int         c;
@@ -530,7 +531,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   oversion = *(int64_t *)pData;
 
   // search table.db
-  TBC     *pTbDbc = NULL;
+  TBC *    pTbDbc = NULL;
   SDecoder dc1 = {0};
   SDecoder dc2 = {0};
 
@@ -554,7 +555,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   metaDecodeEntry(&dc2, &stbEntry);
 
   SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
-  SSchema        *pColumn = NULL;
+  SSchema *       pColumn = NULL;
   int32_t         iCol = 0;
   for (;;) {
     pColumn = NULL;
@@ -584,8 +585,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
     memcpy((void *)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
   } else {
     const STag *pOldTag = (const STag *)ctbEntry.ctbEntry.pTags;
-    STag       *pNewTag = NULL;
-    SArray     *pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal));
+    STag *      pNewTag = NULL;
+    SArray *    pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal));
     if (!pTagArray) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       goto _err;
@@ -669,8 +670,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq, STableMeta
 
 static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   STbDbKey tbDbKey;
-  void    *pKey = NULL;
-  void    *pVal = NULL;
+  void *   pKey = NULL;
+  void *   pVal = NULL;
   int      kLen = 0;
   int      vLen = 0;
   SEncoder coder = {0};
@@ -753,16 +754,11 @@ static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) {
 
 int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_t nTagData, int8_t type, tb_uid_t uid,
                         STagIdxKey **ppTagIdxKey, int32_t *nTagIdxKey) {
-  // int32_t nTagData = 0;
-
-  // if (pTagData) {
-  //   if (IS_VAR_DATA_TYPE(type)) {
-  //     nTagData = varDataTLen(pTagData);
-  //   } else {
-  //     nTagData = tDataTypes[type].bytes;
-  //   }
-  // }
-  *nTagIdxKey = sizeof(STagIdxKey) + nTagData + sizeof(tb_uid_t);
+  if (IS_VAR_DATA_TYPE(type)) {
+    *nTagIdxKey = sizeof(STagIdxKey) + nTagData + VARSTR_HEADER_SIZE + sizeof(tb_uid_t);
+  } else {
+    *nTagIdxKey = sizeof(STagIdxKey) + nTagData + sizeof(tb_uid_t);
+  }
 
   *ppTagIdxKey = (STagIdxKey *)taosMemoryMalloc(*nTagIdxKey);
   if (*ppTagIdxKey == NULL) {
@@ -774,8 +770,16 @@ int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_
   (*ppTagIdxKey)->cid = cid;
   (*ppTagIdxKey)->isNull = (pTagData == NULL) ? 1 : 0;
   (*ppTagIdxKey)->type = type;
-  if (nTagData) memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
-  *(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid;
+
+  // refactor
+  if (IS_VAR_DATA_TYPE(type)) {
+    memcpy((*ppTagIdxKey)->data, (uint16_t *)&nTagData, VARSTR_HEADER_SIZE);
+    memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData);
+    *(tb_uid_t *)((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE + nTagData) = uid;
+  } else {
+    memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
+    *(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid;
+  }
 
   return 0;
 }
@@ -785,14 +789,14 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) {
 }
 
 static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
-  void          *pData = NULL;
+  void *         pData = NULL;
   int            nData = 0;
   STbDbKey       tbDbKey = {0};
   SMetaEntry     stbEntry = {0};
-  STagIdxKey    *pTagIdxKey = NULL;
+  STagIdxKey *   pTagIdxKey = NULL;
   int32_t        nTagIdxKey;
   const SSchema *pTagColumn;       // = &stbEntry.stbEntry.schema.pSchema[0];
-  const void    *pTagData = NULL;  //
+  const void *   pTagData = NULL;  //
   int32_t        nTagData = 0;
   SDecoder       dc = {0};
 
@@ -849,7 +853,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
 
 static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
   SEncoder              coder = {0};
-  void                 *pVal = NULL;
+  void *                pVal = NULL;
   int                   vLen = 0;
   int                   rcode = 0;
   SSkmDbKey             skmDbKey = {0};

source/libs/function/src/builtins.c

@@ -846,7 +846,7 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
   for (int32_t i = 0; i < numOfParams; ++i) {
     SNode*  pPara = nodesListGetNode(pFunc->pParameterList, i);
     uint8_t paraType = ((SExprNode*)pPara)->resType.type;
-    if (!IS_VAR_DATA_TYPE(paraType) && TSDB_DATA_TYPE_NULL != paraType) {
+    if (!IS_VAR_DATA_TYPE(paraType) && !IS_NULL_TYPE(paraType)) {
       return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
     }
     if (TSDB_DATA_TYPE_NCHAR == paraType) {
@@ -859,6 +859,12 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
     uint8_t paraType = ((SExprNode*)pPara)->resType.type;
     int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
     int32_t factor = 1;
+    if (IS_NULL_TYPE(paraType)) {
+      resultType = TSDB_DATA_TYPE_VARCHAR;
+      resultBytes = 0;
+      sepBytes = 0;
+      break;
+    }
     if (TSDB_DATA_TYPE_NCHAR == resultType && TSDB_DATA_TYPE_VARCHAR == paraType) {
       factor *= TSDB_NCHAR_SIZE;
     }

source/libs/index/src/indexFilter.c

@@ -265,23 +265,24 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu
 typedef int (*Filter)(void *a, void *b, int16_t dtype);
 
 int sifGreaterThan(void *a, void *b, int16_t dtype) {
-  __compar_fn_t func = indexGetCompar(dtype);
+  __compar_fn_t func = getComparFunc(dtype, 0);
   return tDoCompare(func, QUERY_GREATER_THAN, a, b);
 }
 int sifGreaterEqual(void *a, void *b, int16_t dtype) {
-  __compar_fn_t func = indexGetCompar(dtype);
+  __compar_fn_t func = getComparFunc(dtype, 0);
   return tDoCompare(func, QUERY_GREATER_EQUAL, a, b);
 }
 int sifLessEqual(void *a, void *b, int16_t dtype) {
-  __compar_fn_t func = indexGetCompar(dtype);
+  __compar_fn_t func = getComparFunc(dtype, 0);
   return tDoCompare(func, QUERY_LESS_EQUAL, a, b);
 }
 int sifLessThan(void *a, void *b, int16_t dtype) {
-  __compar_fn_t func = indexGetCompar(dtype);
+  __compar_fn_t func = getComparFunc(dtype, 0);
   return (int)tDoCompare(func, QUERY_LESS_THAN, a, b);
 }
 int sifEqual(void *a, void *b, int16_t dtype) {
-  __compar_fn_t func = indexGetCompar(dtype);
+  __compar_fn_t func = getComparFunc(dtype, 0);
+  //__compar_fn_t func = indexGetCompar(dtype);
   return (int)tDoCompare(func, QUERY_TERM, a, b);
 }
 static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) {
@@ -397,38 +398,62 @@ static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output)
   return TSDB_CODE_QRY_INVALID_INPUT;
 }
 
-static sif_func_t sifGetOperFn(int32_t funcId) {
+static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *status) {
   // impl later
+  *status = SFLT_ACCURATE_INDEX;
   switch (funcId) {
     case OP_TYPE_GREATER_THAN:
-      return sifGreaterThanFunc;
+      *func = sifGreaterThanFunc;
+      return 0;
     case OP_TYPE_GREATER_EQUAL:
-      return sifGreaterEqualFunc;
+      *func = sifGreaterEqualFunc;
+      return 0;
     case OP_TYPE_LOWER_THAN:
-      return sifLessThanFunc;
+      *func = sifLessThanFunc;
+      return 0;
     case OP_TYPE_LOWER_EQUAL:
-      return sifLessEqualFunc;
+      *func = sifLessEqualFunc;
+      return 0;
     case OP_TYPE_EQUAL:
-      return sifEqualFunc;
+      *func = sifEqualFunc;
+      return 0;
     case OP_TYPE_NOT_EQUAL:
-      return sifNotEqualFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifNotEqualFunc;
+      return 0;
     case OP_TYPE_IN:
-      return sifInFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifInFunc;
+      return 0;
     case OP_TYPE_NOT_IN:
-      return sifNotInFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifNotInFunc;
+      return 0;
     case OP_TYPE_LIKE:
-      return sifLikeFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifLikeFunc;
+      return 0;
     case OP_TYPE_NOT_LIKE:
-      return sifNotLikeFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifNotLikeFunc;
+      return 0;
     case OP_TYPE_MATCH:
-      return sifMatchFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifMatchFunc;
+      return 0;
     case OP_TYPE_NMATCH:
-      return sifNotMatchFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifNotMatchFunc;
+      return 0;
     default:
-      return sifNullFunc;
+      *status = SFLT_NOT_INDEX;
+      *func = sifNullFunc;
+      return 0;
   }
-  return sifNullFunc;
+  return 0;
 }
+// typedef struct filterFuncDict {
+
 static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
   int32_t code = 0;
   int32_t nParam = sifGetOperParamNum(node->opType);
@@ -441,18 +466,14 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
 
   // ugly code, refactor later
   output->arg = ctx->arg;
-  sif_func_t operFn = sifGetOperFn(node->opType);
-  if (ctx->noExec && operFn == NULL) {
-    output->status = SFLT_NOT_INDEX;
-  } else {
-    output->status = SFLT_ACCURATE_INDEX;
-  }
 
+  sif_func_t operFn = sifNullFunc;
+  code = sifGetOperFn(node->opType, &operFn, &output->status);
   if (ctx->noExec) {
     SIF_RET(code);
+  } else {
+    return operFn(&params[0], nParam > 1 ? &params[1] : NULL, output);
   }
-
-  return operFn(&params[0], nParam > 1 ? &params[1] : NULL, output);
 _return:
   taosMemoryFree(params);
   SIF_RET(code);
@@ -477,7 +498,7 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou
       } else if (node->condType == LOGIC_COND_TYPE_OR) {
         taosArrayAddAll(output->result, params[m].result);
       } else if (node->condType == LOGIC_COND_TYPE_NOT) {
-        taosArrayAddAll(output->result, params[m].result);
+        // taosArrayAddAll(output->result, params[m].result);
       }
     }
   } else {

source/libs/qworker/test/qworkerTests.cpp

@@ -172,8 +172,7 @@ int32_t qwtPutReqToFetchQueue(void *node, struct SRpcMsg *pMsg) {
   return 0;
 }
 
-
-int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) {
+int32_t qwtPutReqToQueue(void *node, EQueueType qtype, struct SRpcMsg *pMsg) {
   taosWLockLatch(&qwtTestQueryQueueLock);
   struct SRpcMsg *newMsg = (struct SRpcMsg *)taosMemoryCalloc(1, sizeof(struct SRpcMsg));
   memcpy(newMsg, pMsg, sizeof(struct SRpcMsg));
@@ -181,7 +180,7 @@ int32_t qwtPutReqToQueue(void *node, struct SRpcMsg *pMsg) {
   if (qwtTestQueryQueueWIdx >= qwtTestQueryQueueSize) {
     qwtTestQueryQueueWIdx = 0;
   }
-  
+
   qwtTestQueryQueueNum++;
 
   if (qwtTestQueryQueueWIdx == qwtTestQueryQueueRIdx) {
@@ -702,9 +701,9 @@ void *qwtclientThread(void *param) {
 
   while (!qwtTestStop) {
     qwtTestCaseFinished = false;
-    
+
     qwtBuildQueryReqMsg(&queryRpc);
-    qwtPutReqToQueue((void *)0x1, &queryRpc);
+    qwtPutReqToQueue((void *)0x1, QUERY_QUEUE, &queryRpc);
 
     while (!qwtTestCaseFinished) {
       taosUsleep(1);
@@ -874,7 +873,7 @@ TEST(seqTest, normalCase) {
   
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -907,10 +906,10 @@ TEST(seqTest, cancelFirst) {
 
   stubSetStringToPlan();
   stubSetRpcSendResponse();
-  
+
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -944,10 +943,10 @@ TEST(seqTest, randCase) {
   stubSetCreateExecTask();
 
   taosSeedRand(taosGetTimestampSec());
-  
+
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -1015,10 +1014,10 @@ TEST(seqTest, multithreadRand) {
   stubSetGetDataBlock();
 
   taosSeedRand(taosGetTimestampSec());
-  
+
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -1081,7 +1080,7 @@ TEST(rcTest, shortExecshortDelay) {
 
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -1165,7 +1164,7 @@ TEST(rcTest, longExecshortDelay) {
 
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -1251,7 +1250,7 @@ TEST(rcTest, shortExeclongDelay) {
 
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 
@@ -1332,10 +1331,10 @@ TEST(rcTest, dropTest) {
   stubSetGetDataBlock();
 
   taosSeedRand(taosGetTimestampSec());
-  
+
   SMsgCb msgCb = {0};
   msgCb.mgmt = (void *)mockPointer;
-  msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)qwtPutReqToQueue;
+  msgCb.putToQueueFp = (PutToQueueFp)qwtPutReqToQueue;
   code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, &msgCb);
   ASSERT_EQ(code, 0);
 

source/libs/scalar/src/sclfunc.c

@@ -449,7 +449,7 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
     bool hasNull = false;
     for (int32_t i = 0; i < inputNum; ++i) {
       if (colDataIsNull_s(pInputData[i], k) ||
-          GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_NULL) {
+          IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) {
         colDataAppendNULL(pOutputData, k);
         hasNull = true;
         break;
@@ -526,7 +526,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 
   for (int32_t k = 0; k < numOfRows; ++k) {
     if (colDataIsNull_s(pInputData[0], k) ||
-        GET_PARAM_TYPE(&pInput[0]) == TSDB_DATA_TYPE_NULL) {
+        IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[0]))) {
       colDataAppendNULL(pOutputData, k);
       continue;
     }
@@ -535,7 +535,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
     bool hasNull = false;
     for (int32_t i = 1; i < inputNum; ++i) {
       if (colDataIsNull_s(pInputData[i], k) ||
-          GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_NULL) {
+          IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) {
         hasNull = true;
         break;
       }
@@ -1221,8 +1221,8 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
       if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */
         int32_t ret = convertStringToTimestamp(type, input[k], TSDB_TIME_PRECISION_NANO, &timeVal[k]);
         if (ret != TSDB_CODE_SUCCESS) {
-          colDataAppendNULL(pOutput->columnData, i);
-          continue;
+          hasNull = true;
+          break;
         }
       } else if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_TIMESTAMP) { /* unix timestamp or ts column*/
         GET_TYPED_DATA(timeVal[k], int64_t, type, input[k]);
@@ -1247,8 +1247,8 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
         } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
           timeVal[k] = timeVal[k];
         } else {
-          colDataAppendNULL(pOutput->columnData, i);
-          continue;
+          hasNull = true;
+          break;
         }
       }
     }

source/libs/transport/src/transCli.c

@@ -1151,7 +1151,6 @@ void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) {
     SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
     tDebug("update epset at thread:%d, threadID:%" PRId64 "", i, thrd->thread);
 
-    tsem_t* pSem = pCtx->pSem;
     transSendAsync(thrd->asyncPool, &(cliMsg->q));
   }
 }

tests/script/tsim/db/alter_replica_13.sim

@@ -128,7 +128,6 @@ if $data[0][8] != follower then
 endi
 
 print ============= step5: stop dnode 2
-return
 
 sql select * from db.stb
 if $rows != 1 then