Merge remote-tracking branch 'origin/3.0' into feature/qnode

.clang-format

@@ -86,5 +86,6 @@ SpacesInSquareBrackets: false
 Standard:        Auto
 TabWidth:        8
 UseTab:          Never
+AlignConsecutiveDeclarations: true
 ...
 

contrib/test/craft/clear.sh

 #!/bin/bash
 
 rm -rf 127.0.0.1*
+rm -rf ./data

contrib/test/craft/common.h

+#ifndef TDENGINE_COMMON_H
+#define TDENGINE_COMMON_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+
+#define MAX_PEERS 10
+#define COMMAND_LEN 512
+#define TOKEN_LEN 128
+#define DIR_LEN 256
+#define HOST_LEN 64
+#define ADDRESS_LEN (HOST_LEN + 16)
+
+typedef struct {
+    char host[HOST_LEN];
+    uint32_t port;
+} Addr;
+
+typedef struct {
+    Addr me;
+    Addr peers[MAX_PEERS];
+    int peersCount;
+    char dir[DIR_LEN];
+    char dataDir[DIR_LEN + HOST_LEN * 2];
+} SRaftServerConfig;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // TDENGINE_COMMON_H

contrib/test/craft/raftMain.c

+#include <stdio.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <string.h>
+#include <assert.h>
+#include <getopt.h>
+#include <time.h>
+#include <stdlib.h>
+#include <getopt.h>
+#include <raft.h>
+#include <raft/uv.h>
+#include "raftServer.h"
+#include "common.h"
+
+const char *exe_name;
+
+void parseAddr(const char *addr, char *host, int len, uint32_t *port) {
+	char* tmp = (char*)malloc(strlen(addr) + 1);
+	strcpy(tmp, addr);
+
+	char* context;
+	char* separator = ":";
+	char* token = strtok_r(tmp, separator, &context);
+	if (token) {
+		snprintf(host, len, "%s", token);
+	}
+
+	token = strtok_r(NULL, separator, &context);
+	if (token) {
+		sscanf(token, "%u", port);
+	}
+
+	free(tmp);
+}
+
+// only parse 3 tokens
+int parseCommand(const char* str, char* token1, char* token2, char* token3, int len)
+{
+    char* tmp = (char*)malloc(strlen(str) + 1);
+    strcpy(tmp, str);
+
+    char* context;
+    char* separator = " ";
+    int n = 0;
+
+    char* token = strtok_r(tmp, separator, &context);
+    if (!token) {
+ 	   goto ret;
+    }
+    if (strcmp(token, "") != 0) {
+    	strncpy(token1, token, len);
+    	n++;
+    }
+
+    token = strtok_r(NULL, separator, &context);
+    if (!token) {
+    	goto ret;
+   	}
+    if (strcmp(token, "") != 0) {
+    	strncpy(token2, token, len);
+    	n++;
+    }
+
+    token = strtok_r(NULL, separator, &context);
+    if (!token) {
+    	goto ret;
+    }
+    if (strcmp(token, "") != 0) {
+    	strncpy(token3, token, len);
+    	n++;
+    }
+
+ret:
+    return n;
+    free(tmp);
+}
+
+void *startServerFunc(void *param) {
+	SRaftServer *pServer = (SRaftServer*)param;
+	int32_t r = raftServerStart(pServer);
+	assert(r == 0);
+
+	return NULL;
+}
+
+// Console ---------------------------------
+const char* state2String(unsigned short state) {
+    if (state == RAFT_UNAVAILABLE) {
+        return "RAFT_UNAVAILABLE";
+
+    } else if (state == RAFT_FOLLOWER) {
+        return "RAFT_FOLLOWER";
+
+    } else if (state == RAFT_CANDIDATE) {
+        return "RAFT_CANDIDATE";
+
+    } else if (state == RAFT_LEADER) {
+        return "RAFT_LEADER";
+
+    }
+    return "UNKNOWN_RAFT_STATE";
+}
+
+void printRaftConfiguration(struct raft_configuration *c) {
+	printf("configuration: \n");
+	for (int i = 0; i < c->n; ++i) {
+		printf("%llu -- %d -- %s\n", c->servers->id, c->servers->role, c->servers->address);
+	}
+}
+
+void printRaftState(struct raft *r) {
+    printf("----Raft State: -----------\n");
+    printf("my_id: %llu \n", r->id);
+    printf("address: %s \n", r->address);
+    printf("current_term: %llu \n", r->current_term);
+    printf("voted_for: %llu \n", r->voted_for);
+    printf("role: %s \n", state2String(r->state));
+    printf("commit_index: %llu \n", r->commit_index);
+    printf("last_applied: %llu \n", r->last_applied);
+    printf("last_stored: %llu \n", r->last_stored);
+
+	/*
+    printf("configuration_index: %llu \n", r->configuration_index);
+    printf("configuration_uncommitted_index: %llu \n", r->configuration_uncommitted_index);
+	printRaftConfiguration(&r->configuration);
+	*/
+
+	printf("----------------------------\n");
+}
+
+void putValueCb(struct raft_apply *req, int status, void *result) {
+    raft_free(req);
+	struct raft *r = req->data;
+	if (status != 0) {
+		printf("putValueCb: %s \n", raft_errmsg(r));
+	} else {
+		printf("putValueCb: %s \n", "ok");
+	}
+}
+
+void putValue(struct raft *r, const char *value) {
+    struct raft_buffer buf;
+
+    buf.len = TOKEN_LEN;;
+    buf.base = raft_malloc(buf.len);
+	snprintf(buf.base, buf.len, "%s", value);
+
+    struct raft_apply *req = raft_malloc(sizeof(struct raft_apply));
+	req->data = r;
+    int ret = raft_apply(r, req, &buf, 1, putValueCb);
+	if (ret == 0) {
+    	printf("put %s \n", (char*)buf.base);
+	} else {
+		printf("put error: %s \n", raft_errmsg(r));
+	}
+}
+
+void getValue(const char *key) {
+	char *ptr = getKV(key);
+	if (ptr) {
+		printf("get value: [%s] \n", ptr);
+	} else {
+		printf("value not found for key: [%s] \n", key);
+	}
+}
+
+void console(SRaftServer *pRaftServer) {
+	while (1) {
+		char cmd_buf[COMMAND_LEN];
+		memset(cmd_buf, 0, sizeof(cmd_buf));
+		char *ret = fgets(cmd_buf, COMMAND_LEN, stdin);
+		if (!ret) {
+			exit(-1);
+		}
+
+		int pos = strlen(cmd_buf);                  
+		if(cmd_buf[pos - 1] == '\n') {
+			cmd_buf[pos - 1] = '\0';
+		}
+
+		if (strncmp(cmd_buf, "", COMMAND_LEN) == 0) {
+			continue;
+		}
+
+		char cmd[TOKEN_LEN];
+		memset(cmd, 0, sizeof(cmd));
+
+		char param1[TOKEN_LEN];
+		memset(param1, 0, sizeof(param1));
+
+		char param2[TOKEN_LEN];
+		memset(param2, 0, sizeof(param2));
+
+		parseCommand(cmd_buf, cmd, param1, param2, TOKEN_LEN);
+		if (strcmp(cmd, "addnode") == 0) {
+			printf("not support \n");
+
+			/*
+			char host[HOST_LEN];
+			uint32_t port;
+			parseAddr(param1, host, HOST_LEN, &port);
+			uint64_t rid = raftId(host, port);
+
+			struct raft_change *req = raft_malloc(sizeof(*req));
+			int r = raft_add(&pRaftServer->raft, req, rid, param1, NULL);
+			if (r != 0) {
+				printf("raft_add: %s \n", raft_errmsg(&pRaftServer->raft));
+			}
+			printf("add node: %lu %s \n", rid, param1);
+
+			struct raft_change *req2 = raft_malloc(sizeof(*req2));
+			r = raft_assign(&pRaftServer->raft, req2, rid, RAFT_VOTER, NULL);
+			if (r != 0) {
+				printf("raft_assign: %s \n", raft_errmsg(&pRaftServer->raft));
+			}
+			*/
+
+		} else if (strcmp(cmd, "dropnode") == 0) {
+			printf("not support \n");
+
+		} else if (strcmp(cmd, "put") == 0) {
+			char buf[256];
+			snprintf(buf, sizeof(buf), "%s--%s", param1, param2);
+			putValue(&pRaftServer->raft, buf);
+
+		} else if (strcmp(cmd, "get") == 0) {
+			getValue(param1);
+
+		} else if (strcmp(cmd, "state") == 0) {
+			printRaftState(&pRaftServer->raft);
+
+		} else if (strcmp(cmd, "snapshot") == 0) {
+			printf("not support \n");
+
+		} else if (strcmp(cmd, "help") == 0) {
+			printf("addnode \"127.0.0.1:8888\" \n");
+			printf("dropnode \"127.0.0.1:8888\" \n");
+			printf("put key value \n");
+			printf("get key \n");
+			printf("state \n");
+
+		} else {
+			printf("unknown command: [%s], type \"help\" to see help \n", cmd);
+		}
+
+		//printf("cmd_buf: [%s] \n", cmd_buf);
+	}
+}
+
+void *startConsoleFunc(void *param) {
+	SRaftServer *pServer = (SRaftServer*)param;
+	console(pServer);
+	return NULL;
+}
+
+// Config ---------------------------------
+void usage() {
+	printf("\nusage: \n");
+	printf("%s --me=127.0.0.1:10000 --dir=./data \n", exe_name);
+	printf("\n");
+	printf("%s --me=127.0.0.1:10000 --peers=127.0.0.1:10001,127.0.0.1:10002 --dir=./data \n", exe_name);
+	printf("%s --me=127.0.0.1:10001 --peers=127.0.0.1:10000,127.0.0.1:10002 --dir=./data \n", exe_name);
+	printf("%s --me=127.0.0.1:10002 --peers=127.0.0.1:10000,127.0.0.1:10001 --dir=./data \n", exe_name);
+	printf("\n");
+}
+
+void parseConf(int argc, char **argv, SRaftServerConfig *pConf) {
+	memset(pConf, 0, sizeof(*pConf));
+
+    int option_index, option_value;
+    option_index = 0;
+    static struct option long_options[] = {
+        {"help", no_argument, NULL, 'h'},
+        {"peers", required_argument, NULL, 'p'},
+        {"me", required_argument, NULL, 'm'},
+        {"dir", required_argument, NULL, 'd'},
+        {NULL, 0, NULL, 0}
+    };
+
+    while ((option_value = getopt_long(argc, argv, "hp:m:d:", long_options, &option_index)) != -1) {
+        switch (option_value) {
+        	case 'm': {
+				parseAddr(optarg, pConf->me.host, sizeof(pConf->me.host), &pConf->me.port);
+				break;
+        	}
+
+			case 'p': {
+				char tokens[MAX_PEERS][MAX_TOKEN_LEN];
+				int peerCount = splitString(optarg, ",", tokens, MAX_PEERS);
+				pConf->peersCount = peerCount;
+				for (int i = 0; i < peerCount; ++i) {
+					Addr *pAddr = &pConf->peers[i];
+					parseAddr(tokens[i], pAddr->host, sizeof(pAddr->host), &pAddr->port);
+				}
+				break;
+			}
+
+
+			case 'd': {
+				snprintf(pConf->dir, sizeof(pConf->dir), "%s", optarg);
+				break;
+			}
+
+			case 'h': {
+				usage();
+				exit(-1);
+			}
+
+			default: {
+				usage();
+				exit(-1);
+			}
+		}
+	}
+	snprintf(pConf->dataDir, sizeof(pConf->dataDir), "%s/%s:%u", pConf->dir, pConf->me.host, pConf->me.port);
+}
+
+void printConf(SRaftServerConfig *pConf) {
+	printf("\nconf: \n");
+	printf("me: %s:%u \n", pConf->me.host, pConf->me.port);
+	printf("peersCount: %d \n", pConf->peersCount);
+	for (int i = 0; i < pConf->peersCount; ++i) {
+		Addr *pAddr = &pConf->peers[i];
+		printf("peer%d: %s:%u \n", i, pAddr->host, pAddr->port);
+	}
+	printf("dataDir: %s \n\n", pConf->dataDir);
+
+}
+
+
+int main(int argc, char **argv) { 
+	srand(time(NULL));
+	int32_t ret;
+
+	exe_name = argv[0];
+	if (argc < 3) {
+		usage();
+		exit(-1);
+	}
+
+	SRaftServerConfig conf;
+	parseConf(argc, argv, &conf);
+	printConf(&conf);	
+
+	char cmd_buf[COMMAND_LEN];
+    snprintf(cmd_buf, sizeof(cmd_buf), "mkdir -p %s", conf.dataDir);
+    system(cmd_buf);
+
+	struct raft_fsm fsm;
+	initFsm(&fsm);
+
+	SRaftServer raftServer;
+	ret = raftServerInit(&raftServer, &conf, &fsm);
+	assert(ret == 0);
+
+	pthread_t tidRaftServer;
+	pthread_create(&tidRaftServer, NULL, startServerFunc, &raftServer);
+
+	pthread_t tidConsole;
+	pthread_create(&tidConsole, NULL, startConsoleFunc, &raftServer);
+
+	while (1) {
+		sleep(10);
+	}
+
+	return 0;
+}

contrib/test/craft/raftServer.c

+#include <stdlib.h>
+#include "common.h"
+#include "raftServer.h"
+
+char *keys;
+char *values;
+
+void initStore() {
+	keys = malloc(MAX_RECORD_COUNT * MAX_KV_LEN);
+	values = malloc(MAX_RECORD_COUNT * MAX_KV_LEN);
+	writeIndex = 0;
+}
+
+void destroyStore() {
+	free(keys);
+	free(values);
+}
+
+void putKV(const char *key, const char *value) {
+	if (writeIndex < MAX_RECORD_COUNT) {
+		strncpy(&keys[writeIndex], key, MAX_KV_LEN);
+		strncpy(&values[writeIndex], value, MAX_KV_LEN);
+		writeIndex++;
+	}
+}
+
+char *getKV(const char *key) {
+	for (int i = 0; i < MAX_RECORD_COUNT; ++i) {
+		if (strcmp(&keys[i], key) == 0) {
+			return &values[i];
+		}
+	}
+	return NULL;
+}
+
+
+int splitString(const char* str, char* separator, char (*arr)[MAX_TOKEN_LEN], int n_arr)
+{
+    if (n_arr <= 0) {
+    	return -1;
+    }
+
+    char* tmp = (char*)malloc(strlen(str) + 1);
+    strcpy(tmp, str);
+    char* context;
+    int n = 0;
+
+    char* token = strtok_r(tmp, separator, &context);
+    if (!token) {
+    	goto ret;
+    }
+    strncpy(arr[n], token, MAX_TOKEN_LEN);
+    n++;
+
+    while (1) {
+    token = strtok_r(NULL, separator, &context);
+    	if (!token || n >= n_arr) {
+    	    goto ret;
+    	}
+    	strncpy(arr[n], token, MAX_TOKEN_LEN);
+    	n++;
+    }
+
+ret:
+    free(tmp);
+    return n;
+}
+
+uint64_t raftId(const char *host, uint32_t port) {
+    uint32_t host_uint32 = (uint32_t)inet_addr(host);
+    assert(host_uint32 != (uint32_t)-1);
+    uint64_t code = ((uint64_t)host_uint32) << 32 | port;
+	return code;
+}
+
+int32_t raftServerInit(SRaftServer *pRaftServer, const SRaftServerConfig *pConf, struct raft_fsm *pFsm) {
+	int ret;
+
+    snprintf(pRaftServer->host, sizeof(pRaftServer->host), "%s", pConf->me.host);
+	pRaftServer->port = pConf->me.port;
+    snprintf(pRaftServer->address, sizeof(pRaftServer->address), "%s:%u", pRaftServer->host, pRaftServer->port);
+	strncpy(pRaftServer->dir, pConf->dataDir, sizeof(pRaftServer->dir));
+
+	pRaftServer->raftId = raftId(pRaftServer->host, pRaftServer->port);
+	pRaftServer->fsm = pFsm;
+
+	ret = uv_loop_init(&pRaftServer->loop);
+	if (!ret) {
+		fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
+	}
+
+	ret = raft_uv_tcp_init(&pRaftServer->transport, &pRaftServer->loop);
+	if (!ret) {
+		fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
+	}
+
+	ret = raft_uv_init(&pRaftServer->io, &pRaftServer->loop, pRaftServer->dir, &pRaftServer->transport);
+	if (!ret) {
+		fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
+	}
+
+	ret = raft_init(&pRaftServer->raft, &pRaftServer->io, pRaftServer->fsm, pRaftServer->raftId, pRaftServer->address);
+	if (!ret) {
+		fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
+	}
+
+    struct raft_configuration conf;
+    raft_configuration_init(&conf);
+    raft_configuration_add(&conf, pRaftServer->raftId, pRaftServer->address, RAFT_VOTER);
+	printf("add myself: %llu - %s \n", pRaftServer->raftId, pRaftServer->address);
+	for (int i = 0; i < pConf->peersCount; ++i) {
+		const Addr *pAddr = &pConf->peers[i];
+		raft_id rid = raftId(pAddr->host, pAddr->port);
+		char addrBuf[ADDRESS_LEN];
+		snprintf(addrBuf, sizeof(addrBuf), "%s:%u", pAddr->host, pAddr->port);
+		raft_configuration_add(&conf, rid, addrBuf, RAFT_VOTER);
+		printf("add peers: %llu - %s \n", rid, addrBuf);
+	}
+
+    raft_bootstrap(&pRaftServer->raft, &conf);
+
+	return 0;
+}
+
+int32_t raftServerStart(SRaftServer *pRaftServer) {
+	int ret;
+	ret = raft_start(&pRaftServer->raft);
+	if (!ret) {
+		fprintf(stderr, "%s \n", raft_errmsg(&pRaftServer->raft));
+	}
+
+	uv_run(&pRaftServer->loop, UV_RUN_DEFAULT);
+}
+
+
+void raftServerClose(SRaftServer *pRaftServer) {
+
+}
+
+
+int fsmApplyCb(struct raft_fsm *pFsm, const struct raft_buffer *buf, void **result) {
+    char *msg = (char*)buf->base;
+    printf("fsm apply: %s \n", msg);
+
+	char arr[2][MAX_TOKEN_LEN];
+	splitString(msg, "--", arr, 2);
+	putKV(arr[0], arr[1]);
+
+    return 0;
+}
+
+int32_t initFsm(struct raft_fsm *fsm) {
+	initStore();
+	fsm->apply = fsmApplyCb;
+	return 0;
+}

contrib/test/craft/raftServer.h

+#ifndef TDENGINE_RAFT_SERVER_H
+#define TDENGINE_RAFT_SERVER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <assert.h>
+#include <string.h>
+#include "raft.h"
+#include "raft/uv.h"
+#include "common.h"
+
+
+// simulate a db store, just for test
+#define MAX_KV_LEN 100
+#define MAX_RECORD_COUNT 500
+char *keys;
+char *values;
+int writeIndex;
+
+void initStore();
+void destroyStore();
+void putKV(const char *key, const char *value);
+char *getKV(const char *key);
+
+typedef struct {
+	char dir[DIR_LEN + HOST_LEN * 2];  /* Data dir of UV I/O backend */
+	char host[HOST_LEN];
+    uint32_t port;
+	char address[ADDRESS_LEN];         /* Raft instance address */
+	raft_id raftId;                    /* For vote */
+	struct raft_fsm *fsm;              /* Sample application FSM */
+
+	struct raft raft;                   /* Raft instance */
+	struct raft_io io;                  /* UV I/O backend */
+	struct uv_loop_s loop;              /* UV loop */
+	struct raft_uv_transport transport; /* UV I/O backend transport */
+} SRaftServer;
+
+#define MAX_TOKEN_LEN 32
+int splitString(const char* str, char* separator, char (*arr)[MAX_TOKEN_LEN], int n_arr);
+
+uint64_t raftId(const char *host, uint32_t port);
+int32_t raftServerInit(SRaftServer *pRaftServer, const SRaftServerConfig *pConf, struct raft_fsm *pFsm);
+int32_t raftServerStart(SRaftServer *pRaftServer);
+void raftServerClose(SRaftServer *pRaftServer);
+
+
+int initFsm(struct raft_fsm *fsm);
+
+
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // TDENGINE_RAFT_SERVER_H

include/common/taosmsg.h

@@ -918,18 +918,15 @@ typedef struct SShowRsp {
 
 typedef struct {
   char    ep[TSDB_EP_LEN];  // end point, hostname:port
-  int32_t reserve[8];
 } SCreateDnodeMsg;
 
 typedef struct {
   int32_t dnodeId;
-  int32_t reserve[8];
 } SDropDnodeMsg;
 
 typedef struct {
   int32_t dnodeId;
   char    config[TSDB_DNODE_CONFIG_LEN];
-  int32_t reserve[8];
 } SCfgDnodeMsg;
 
 typedef struct {
@@ -938,7 +935,6 @@ typedef struct {
 
 typedef struct {
   int32_t  dnodeId;
-  int8_t   align[3];
   int8_t   replica;
   SReplica replicas[TSDB_MAX_REPLICA];
 } SCreateMnodeInMsg, SAlterMnodeInMsg;

include/libs/parser/parsenodes.h

@@ -160,6 +160,13 @@ typedef struct SInsertStmtInfo {
   const char* sql;                 // current sql statement position
 } SInsertStmtInfo;
 
+typedef struct SDclStmtInfo {
+  int16_t     nodeType;
+  int16_t     msgType;
+  char*       pMsg;
+  int32_t     msgLen;
+} SDclStmtInfo;
+
 #ifdef __cplusplus
 }
 #endif

include/libs/parser/parser.h

@@ -22,20 +22,11 @@ extern "C" {
 
 #include "parsenodes.h"
 
-/**
- * True will be returned if the input sql string is insert, false otherwise.
- * @param pStr    sql string
- * @param length  length of the sql string
- * @return
- */
-bool qIsInsertSql(const char* pStr, size_t length);
-
 typedef struct SParseContext {
   SParseBasicCtx  ctx;
   void            *pRpc;
   struct SCatalog *pCatalog;
   const SEpSet    *pEpSet;
-  int64_t          id;             // query id, generated by uuid generator
   int8_t           schemaAttached; // denote if submit block is built with table schema or not
   const char      *pSql;           // sql string
   size_t           sqlLen;         // length of the sql string
@@ -51,17 +42,9 @@ typedef struct SParseContext {
  * @param msg      extended error message if exists.
  * @return         error code
  */
-int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCtx, int32_t* type, void** pOutput, int32_t* outputLen, char* msg, int32_t msgLen);
+int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
 
-/**
- * Parse the insert sql statement.
- * @param pStr            sql string
- * @param length          length of the sql string
- * @param id              operator id, generated by uuid generator.
- * @param msg             extended error message if exists to help avoid the problem in sql statement.
- * @return                data in binary format to submit to vnode directly.
- */
- int32_t qParseInsertSql(SParseContext* pContext, struct SInsertStmtInfo** pInfo);
+bool qIsDclQuery(const SQueryNode* pQuery);
 
 /**
  * Convert a normal sql statement to only query tags information to enable that the subscribe client can be aware quickly of the true vgroup ids that

source/client/src/clientImpl.c

@@ -144,72 +144,66 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
 
   tscDebugL("0x%"PRIx64" SQL: %s", pRequest->requestId, pRequest->sqlstr);
 
-  int32_t code = 0;
-  if (qIsInsertSql(pRequest->sqlstr, sqlLen)) {
-    // todo add
-  } else {
-    int32_t type = 0;
-    void*   output = NULL;
-    int32_t outputLen = 0;
-
-    SParseBasicCtx c = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj)};
-    code = qParseQuerySql(pRequest->sqlstr, sqlLen, &c, &type, &output, &outputLen, pRequest->msgBuf, ERROR_MSG_BUF_DEFAULT_SIZE);
-    if (type == TSDB_MSG_TYPE_CREATE_USER || type == TSDB_MSG_TYPE_SHOW || type == TSDB_MSG_TYPE_DROP_USER ||
-        type == TSDB_MSG_TYPE_DROP_ACCT || type == TSDB_MSG_TYPE_CREATE_DB || type == TSDB_MSG_TYPE_CREATE_ACCT ||
-        type == TSDB_MSG_TYPE_CREATE_TABLE || type == TSDB_MSG_TYPE_CREATE_STB || type == TSDB_MSG_TYPE_USE_DB ||
-        type == TSDB_MSG_TYPE_DROP_DB || type == TSDB_MSG_TYPE_DROP_STB) {
-      pRequest->type = type;
-      pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = output, .len = outputLen};
-
-      SRequestMsgBody body = buildRequestMsgImpl(pRequest);
-      SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
-
-      if (type == TSDB_MSG_TYPE_CREATE_TABLE) {
-        struct SCatalog* pCatalog = NULL;
-
-        char buf[12] = {0};
-        sprintf(buf, "%d", pTscObj->pAppInfo->clusterId);
-        code = catalogGetHandle(buf, &pCatalog);
-        if (code != 0) {
-          pRequest->code = code;
-          return pRequest;
-        }
-
-        SCreateTableMsg* pMsg = body.msgInfo.pMsg;
-
-        SName t = {0};
-        tNameFromString(&t, pMsg->name, T_NAME_ACCT|T_NAME_DB|T_NAME_TABLE);
-
-        char db[TSDB_DB_NAME_LEN + TS_PATH_DELIMITER_LEN + TSDB_ACCT_ID_LEN] = {0};
-        tNameGetFullDbName(&t, db);
-
-        SVgroupInfo info = {0};
-        catalogGetTableHashVgroup(pCatalog, pTscObj->pTransporter, pEpSet, db, tNameGetTableName(&t), &info);
-
-        int64_t transporterId = 0;
-        SEpSet ep = {0};
-        ep.inUse = info.inUse;
-        ep.numOfEps = info.numOfEps;
-        for(int32_t i = 0; i < ep.numOfEps; ++i) {
-          ep.port[i] = info.epAddr[i].port;
-          tstrncpy(ep.fqdn[i], info.epAddr[i].fqdn, tListLen(ep.fqdn[i]));
-        }
-
-        sendMsgToServer(pTscObj->pTransporter, &ep, &body, &transporterId);
-      } else {
-        int64_t transporterId = 0;
-        sendMsgToServer(pTscObj->pTransporter, pEpSet, &body, &transporterId);
+  SParseContext cxt = {
+    .ctx = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj)},
+    .pSql = pRequest->sqlstr,
+    .sqlLen = sqlLen,
+    .pMsg = pRequest->msgBuf,
+    .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
+  };
+  SQueryNode* pQuery = NULL;
+  int32_t code = qParseQuerySql(&cxt, &pQuery);
+  if (qIsDclQuery(pQuery)) {
+    SDclStmtInfo* pDcl = (SDclStmtInfo*)pQuery;
+    pRequest->type = pDcl->msgType;
+    pRequest->body.requestMsg = (SReqMsgInfo){.pMsg = pDcl->pMsg, .len = pDcl->msgLen};
+
+    SRequestMsgBody body = buildRequestMsgImpl(pRequest);
+    SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
+
+    if (pDcl->msgType == TSDB_MSG_TYPE_CREATE_TABLE) {
+      struct SCatalog* pCatalog = NULL;
+
+      char buf[12] = {0};
+      sprintf(buf, "%d", pTscObj->pAppInfo->clusterId);
+      code = catalogGetHandle(buf, &pCatalog);
+      if (code != 0) {
+        pRequest->code = code;
+        return pRequest;
+      }
+
+      SCreateTableMsg* pMsg = body.msgInfo.pMsg;
+
+      SName t = {0};
+      tNameFromString(&t, pMsg->name, T_NAME_ACCT|T_NAME_DB|T_NAME_TABLE);
+
+      char db[TSDB_DB_NAME_LEN + TS_PATH_DELIMITER_LEN + TSDB_ACCT_ID_LEN] = {0};
+      tNameGetFullDbName(&t, db);
+
+      SVgroupInfo info = {0};
+      catalogGetTableHashVgroup(pCatalog, pTscObj->pTransporter, pEpSet, db, tNameGetTableName(&t), &info);
+
+      int64_t transporterId = 0;
+      SEpSet ep = {0};
+      ep.inUse = info.inUse;
+      ep.numOfEps = info.numOfEps;
+      for(int32_t i = 0; i < ep.numOfEps; ++i) {
+        ep.port[i] = info.epAddr[i].port;
+        tstrncpy(ep.fqdn[i], info.epAddr[i].fqdn, tListLen(ep.fqdn[i]));
       }
 
-      tsem_wait(&pRequest->body.rspSem);
-      destroyRequestMsgBody(&body);
+      sendMsgToServer(pTscObj->pTransporter, &ep, &body, &transporterId);
     } else {
-      assert(0);
+      int64_t transporterId = 0;
+      sendMsgToServer(pTscObj->pTransporter, pEpSet, &body, &transporterId);
     }
 
-    tfree(c.db);
+    tsem_wait(&pRequest->body.rspSem);
+    destroyRequestMsgBody(&body);
   }
 
+  tfree(cxt.ctx.db);
+
   if (code != TSDB_CODE_SUCCESS) {
     pRequest->code = code;
     return pRequest;

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

@@ -349,7 +349,7 @@ static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
   SReplica *pReplica = &pOption->replicas[0];
   pReplica->id = 1;
   pReplica->port = pDnode->opt.serverPort;
-  tstrncpy(pReplica->fqdn, pDnode->opt.localFqdn, TSDB_FQDN_LEN);
+  memcpy(pReplica->fqdn, pDnode->opt.localFqdn, TSDB_FQDN_LEN);
 
   SMnodeMgmt *pMgmt = &pDnode->mmgmt;
   pMgmt->selfIndex = pOption->selfIndex;
@@ -376,7 +376,7 @@ static int32_t dndBuildMnodeOptionFromMsg(SDnode *pDnode, SMnodeOpt *pOption, SC
     SReplica *pReplica = &pOption->replicas[i];
     pReplica->id = pMsg->replicas[i].id;
     pReplica->port = pMsg->replicas[i].port;
-    tstrncpy(pReplica->fqdn, pMsg->replicas[i].fqdn, TSDB_FQDN_LEN);
+    memcpy(pReplica->fqdn, pMsg->replicas[i].fqdn, TSDB_FQDN_LEN);
     if (pReplica->id == pOption->dnodeId) {
       pOption->selfIndex = i;
     }
@@ -479,9 +479,11 @@ static int32_t dndDropMnode(SDnode *pDnode) {
     return -1;
   }
 
+  dndReleaseMnode(pDnode, pMnode);
   dndStopMnodeWorker(pDnode);
   dndWriteMnodeFile(pDnode);
   mndClose(pMnode);
+  pMgmt->pMnode = NULL;
   mndDestroy(pDnode->dir.mnode);
 
   return 0;
@@ -499,7 +501,7 @@ static SCreateMnodeInMsg *dndParseCreateMnodeMsg(SRpcMsg *pRpcMsg) {
 }
 
 static int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
-  SCreateMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg->pCont);
+  SCreateMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
 
   if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
     terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
@@ -515,18 +517,23 @@ static int32_t dndProcessCreateMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
 }
 
 static int32_t dndProcessAlterMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
-  SAlterMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg->pCont);
+  SAlterMnodeInMsg *pMsg = dndParseCreateMnodeMsg(pRpcMsg);
 
   if (pMsg->dnodeId != dndGetDnodeId(pDnode)) {
     terrno = TSDB_CODE_DND_MNODE_ID_INVALID;
     return -1;
-  } else {
-    SMnodeOpt option = {0};
-    if (dndBuildMnodeOptionFromMsg(pDnode, &option, pMsg) != 0) {
-      return -1;
-    }
-    return dndAlterMnode(pDnode, &option);
   }
+
+  SMnodeOpt option = {0};
+  if (dndBuildMnodeOptionFromMsg(pDnode, &option, pMsg) != 0) {
+    return -1;
+  }
+
+  if (dndAlterMnode(pDnode, &option) != 0) {
+    return -1;
+  }
+
+  return dndWriteMnodeFile(pDnode);
 }
 
 static int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg) {
@@ -555,16 +562,17 @@ static void dndProcessMnodeMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg) {
       code = dndProcessDropMnodeReq(pDnode, pMsg);
       break;
     default:
-      code = TSDB_CODE_MSG_NOT_PROCESSED;
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      code = -1;
       break;
   }
 
   if (pMsg->msgType & 1u) {
+    if (code != 0) code = terrno;
     SRpcMsg rsp = {.code = code, .handle = pMsg->handle};
     rpcSendResponse(&rsp);
   }
   rpcFreeCont(pMsg->pCont);
-  pMsg->pCont = NULL;
   taosFreeQitem(pMsg);
 }
 
@@ -625,8 +633,6 @@ static void dndProcessMnodeSyncQueue(SDnode *pDnode, SMnodeMsg *pMsg) {
 }
 
 static int32_t dndWriteMnodeMsgToQueue(SMnode *pMnode, taos_queue pQueue, SRpcMsg *pRpcMsg) {
-  assert(pQueue);
-
   SMnodeMsg *pMsg = mndInitMsg(pMnode, pRpcMsg);
   if (pMsg == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -647,15 +653,18 @@ void dndProcessMnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pRpcMsg, SEpSet *pEpSet) {
   SMnode     *pMnode = dndAcquireMnode(pDnode);
 
   SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg));
+  if (pMsg != NULL) *pMsg = *pRpcMsg;
+
   if (pMsg == NULL || taosWriteQitem(pMgmt->pMgmtQ, pMsg) != 0) {
     if (pRpcMsg->msgType & 1u) {
       SRpcMsg rsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_OUT_OF_MEMORY};
       rpcSendResponse(&rsp);
     }
     rpcFreeCont(pRpcMsg->pCont);
-    pRpcMsg->pCont = NULL;
     taosFreeQitem(pMsg);
   }
+
+  dndReleaseMnode(pDnode, pMnode);
 }
 
 void dndProcessMnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
@@ -894,6 +903,11 @@ int32_t dndInitMnode(SDnode *pDnode) {
     return -1;
   }
 
+  if (dndAllocMnodeMgmtQueue(pDnode) != 0) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
   char path[PATH_MAX];
   snprintf(path, PATH_MAX, "%s/mnode.json", pDnode->dir.dnode);
   pMgmt->file = strdup(path);
@@ -935,8 +949,9 @@ void dndCleanupMnode(SDnode *pDnode) {
   SMnodeMgmt *pMgmt = &pDnode->mmgmt;
 
   dInfo("dnode-mnode start to clean up");
-  dndStopMnodeWorker(pDnode);
+  if (pMgmt->pMnode) dndStopMnodeWorker(pDnode);
   dndCleanupMnodeMgmtWorker(pDnode);
+  dndFreeMnodeMgmtQueue(pDnode);
   tfree(pMgmt->file);
   mndClose(pMgmt->pMnode);
   dInfo("dnode-mnode is cleaned up");

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

@@ -140,7 +140,7 @@ static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
   DndMsgFp fp = pMgmt->msgFp[msgType];
   if (fp != NULL) {
     (*fp)(pDnode, pMsg, pEpSet);
-    dTrace("RPC %p, rsp:%s is processed, code:0x%0X", pMsg->handle, taosMsg[msgType], pMsg->code & 0XFFFF);
+    dTrace("RPC %p, rsp:%s is processed, code:0x%x", pMsg->handle, taosMsg[msgType], pMsg->code & 0XFFFF);
   } else {
     dError("RPC %p, rsp:%s not processed", pMsg->handle, taosMsg[msgType]);
     rpcFreeCont(pMsg->pCont);
@@ -188,7 +188,7 @@ static void dndProcessRequest(void *param, SRpcMsg *pMsg, SEpSet *pEpSet) {
 
   int32_t msgType = pMsg->msgType;
   if (msgType == TSDB_MSG_TYPE_NETWORK_TEST) {
-    dTrace("RPC %p, network test req, app:%p will be processed", pMsg->handle, pMsg->ahandle);
+    dTrace("RPC %p, network test req, app:%p will be processed, code:0x%x", pMsg->handle, pMsg->ahandle, pMsg->code);
     dndProcessDnodeReq(pDnode, pMsg, pEpSet);
     return;
   }

source/dnode/mgmt/impl/test/CMakeLists.txt

@@ -7,7 +7,7 @@ add_subdirectory(cluster)
 add_subdirectory(db)
 add_subdirectory(dnode)
 # add_subdirectory(func)
-# add_subdirectory(mnode)
+add_subdirectory(mnode)
 add_subdirectory(profile)
 add_subdirectory(show)
 add_subdirectory(stb)

source/dnode/mgmt/impl/test/mnode/CMakeLists.txt

+aux_source_directory(. MTEST_SRC)
+add_executable(dnode_test_mnode ${MTEST_SRC})
+target_link_libraries(
+    dnode_test_mnode
+    PUBLIC sut
+)
+
+add_test(
+    NAME dnode_test_mnode
+    COMMAND dnode_test_mnode
+)

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

+/**
+ * @file dnode.cpp
+ * @author slguan (slguan@taosdata.com)
+ * @brief DNODE module dnode-msg tests
+ * @version 0.1
+ * @date 2021-12-15
+ *
+ * @copyright Copyright (c) 2021
+ *
+ */
+
+#include "base.h"
+
+class DndTestMnode : public ::testing::Test {
+ public:
+  void SetUp() override {}
+  void TearDown() override {}
+
+ public:
+  static void SetUpTestSuite() {
+    test.Init("/tmp/dnode_test_mnode1", 9061);
+    const char* fqdn = "localhost";
+    const char* firstEp = "localhost:9061";
+
+    server2.Start("/tmp/dnode_test_mnode2", fqdn, 9062, firstEp);
+    server3.Start("/tmp/dnode_test_mnode3", fqdn, 9063, firstEp);
+    server4.Start("/tmp/dnode_test_mnode4", fqdn, 9064, firstEp);
+    server5.Start("/tmp/dnode_test_mnode5", fqdn, 9065, firstEp);
+    taosMsleep(300);
+  }
+
+  static void TearDownTestSuite() {
+    server2.Stop();
+    server3.Stop();
+    server4.Stop();
+    server5.Stop();
+    test.Cleanup();
+  }
+
+  static Testbase   test;
+  static TestServer server2;
+  static TestServer server3;
+  static TestServer server4;
+  static TestServer server5;
+};
+
+Testbase   DndTestMnode::test;
+TestServer DndTestMnode::server2;
+TestServer DndTestMnode::server3;
+TestServer DndTestMnode::server4;
+TestServer DndTestMnode::server5;
+
+TEST_F(DndTestMnode, 01_ShowDnode) {
+  test.SendShowMetaMsg(TSDB_MGMT_TABLE_MNODE, "");
+  CHECK_META("show mnodes", 5);
+
+  CHECK_SCHEMA(0, TSDB_DATA_TYPE_SMALLINT, 2, "id");
+  CHECK_SCHEMA(1, TSDB_DATA_TYPE_BINARY, TSDB_EP_LEN + VARSTR_HEADER_SIZE, "endpoint");
+  CHECK_SCHEMA(2, TSDB_DATA_TYPE_BINARY, 12 + VARSTR_HEADER_SIZE, "role");
+  CHECK_SCHEMA(3, TSDB_DATA_TYPE_TIMESTAMP, 8, "role_time");
+  CHECK_SCHEMA(4, TSDB_DATA_TYPE_TIMESTAMP, 8, "create_time");
+
+  test.SendShowRetrieveMsg();
+  EXPECT_EQ(test.GetShowRows(), 1);
+
+  CheckInt16(1);
+  CheckBinary("localhost:9061", TSDB_EP_LEN);
+  CheckBinary("master", 12);
+  CheckInt64(0);
+  CheckTimestamp();
+}
+
+TEST_F(DndTestMnode, 02_Create_Mnode_Invalid_Id) {
+  {
+    int32_t contLen = sizeof(SCreateMnodeMsg);
+
+    SCreateMnodeMsg* pReq = (SCreateMnodeMsg*)rpcMallocCont(contLen);
+    pReq->dnodeId = htonl(1);
+
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_MNODE, pReq, contLen);
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, TSDB_CODE_MND_MNODE_ALREADY_EXIST);
+  }
+}
+
+TEST_F(DndTestMnode, 03_Create_Mnode_Invalid_Id) {
+  {
+    int32_t contLen = sizeof(SCreateMnodeMsg);
+
+    SCreateMnodeMsg* pReq = (SCreateMnodeMsg*)rpcMallocCont(contLen);
+    pReq->dnodeId = htonl(2);
+
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_MNODE, pReq, contLen);
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, TSDB_CODE_MND_DNODE_NOT_EXIST);
+  }
+}
+
+TEST_F(DndTestMnode, 04_Create_Mnode) {
+  {
+    // create dnode
+    int32_t contLen = sizeof(SCreateDnodeMsg);
+
+    SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
+    strcpy(pReq->ep, "localhost:9062");
+
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DNODE, pReq, contLen);
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, 0);
+
+    taosMsleep(1300);
+    test.SendShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "");
+    test.SendShowRetrieveMsg();
+    EXPECT_EQ(test.GetShowRows(), 2);
+  }
+
+  {
+    // create mnode
+    int32_t contLen = sizeof(SCreateMnodeMsg);
+
+    SCreateMnodeMsg* pReq = (SCreateMnodeMsg*)rpcMallocCont(contLen);
+    pReq->dnodeId = htonl(2);
+
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_MNODE, pReq, contLen);
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, 0);
+
+    test.SendShowMetaMsg(TSDB_MGMT_TABLE_MNODE, "");
+    test.SendShowRetrieveMsg();
+    EXPECT_EQ(test.GetShowRows(), 2);
+
+    CheckInt16(1);
+    CheckInt16(2);
+    CheckBinary("localhost:9061", TSDB_EP_LEN);
+    CheckBinary("localhost:9062", TSDB_EP_LEN);
+    CheckBinary("master", 12);
+    CheckBinary("slave", 12);
+    CheckInt64(0);
+    CheckInt64(0);
+    CheckTimestamp();
+    CheckTimestamp();
+  }
+
+  {
+    // drop mnode
+    int32_t contLen = sizeof(SDropMnodeMsg);
+
+    SDropMnodeMsg* pReq = (SDropMnodeMsg*)rpcMallocCont(contLen);
+    pReq->dnodeId = htonl(2);
+
+    SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_DROP_MNODE, pReq, contLen);
+    ASSERT_NE(pMsg, nullptr);
+    ASSERT_EQ(pMsg->code, 0);
+
+    test.SendShowMetaMsg(TSDB_MGMT_TABLE_MNODE, "");
+    test.SendShowRetrieveMsg();
+    EXPECT_EQ(test.GetShowRows(), 1);
+
+    CheckInt16(1);
+    CheckBinary("localhost:9061", TSDB_EP_LEN);
+    CheckBinary("master", 12);
+    CheckInt64(0);
+    CheckTimestamp();
+  }
+}
+// {
+//   int32_t contLen = sizeof(SDropDnodeMsg);
+
+//   SDropDnodeMsg* pReq = (SDropDnodeMsg*)rpcMallocCont(contLen);
+//   pReq->dnodeId = htonl(2);
+
+//   SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_DROP_DNODE, pReq, contLen);
+//   ASSERT_NE(pMsg, nullptr);
+//   ASSERT_EQ(pMsg->code, 0);
+// }
+
+// test.SendShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "");
+// CHECK_META("show dnodes", 7);
+// test.SendShowRetrieveMsg();
+// EXPECT_EQ(test.GetShowRows(), 1);
+
+// CheckInt16(1);
+// CheckBinary("localhost:9061", TSDB_EP_LEN);
+// CheckInt16(0);
+// CheckInt16(1);
+// CheckBinary("ready", 10);
+// CheckTimestamp();
+// CheckBinary("", 24);
+
+// {
+//   int32_t contLen = sizeof(SCreateDnodeMsg);
+
+//   SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
+//   strcpy(pReq->ep, "localhost:9063");
+
+//   SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DNODE, pReq, contLen);
+//   ASSERT_NE(pMsg, nullptr);
+//   ASSERT_EQ(pMsg->code, 0);
+// }
+
+// {
+//   int32_t contLen = sizeof(SCreateDnodeMsg);
+
+//   SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
+//   strcpy(pReq->ep, "localhost:9064");
+
+//   SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DNODE, pReq, contLen);
+//   ASSERT_NE(pMsg, nullptr);
+//   ASSERT_EQ(pMsg->code, 0);
+// }
+
+// {
+//   int32_t contLen = sizeof(SCreateDnodeMsg);
+
+//   SCreateDnodeMsg* pReq = (SCreateDnodeMsg*)rpcMallocCont(contLen);
+//   strcpy(pReq->ep, "localhost:9065");
+
+//   SRpcMsg* pMsg = test.SendMsg(TSDB_MSG_TYPE_CREATE_DNODE, pReq, contLen);
+//   ASSERT_NE(pMsg, nullptr);
+//   ASSERT_EQ(pMsg->code, 0);
+// }
+
+// taosMsleep(1300);
+// test.SendShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "");
+// CHECK_META("show dnodes", 7);
+// test.SendShowRetrieveMsg();
+// EXPECT_EQ(test.GetShowRows(), 4);
+
+// CheckInt16(1);
+// CheckInt16(3);
+// CheckInt16(4);
+// CheckInt16(5);
+// CheckBinary("localhost:9061", TSDB_EP_LEN);
+// CheckBinary("localhost:9063", TSDB_EP_LEN);
+// CheckBinary("localhost:9064", TSDB_EP_LEN);
+// CheckBinary("localhost:9065", TSDB_EP_LEN);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+
+// // restart
+// uInfo("stop all server");
+// test.Restart();
+// server2.Restart();
+// server3.Restart();
+// server4.Restart();
+// server5.Restart();
+
+// taosMsleep(1300);
+// test.SendShowMetaMsg(TSDB_MGMT_TABLE_DNODE, "");
+// CHECK_META("show dnodes", 7);
+// test.SendShowRetrieveMsg();
+// EXPECT_EQ(test.GetShowRows(), 4);
+
+// CheckInt16(1);
+// CheckInt16(3);
+// CheckInt16(4);
+// CheckInt16(5);
+// CheckBinary("localhost:9061", TSDB_EP_LEN);
+// CheckBinary("localhost:9063", TSDB_EP_LEN);
+// CheckBinary("localhost:9064", TSDB_EP_LEN);
+// CheckBinary("localhost:9065", TSDB_EP_LEN);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(0);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckInt16(1);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckBinary("ready", 10);
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckTimestamp();
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+// CheckBinary("", 24);
+// }

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

@@ -290,11 +290,6 @@ typedef struct SMnodeMsg {
   char    db[TSDB_FULL_DB_NAME_LEN];
   int32_t acctId;
   SMnode *pMnode;
-  int16_t received;
-  int16_t successed;
-  int16_t expected;
-  int16_t retry;
-  int32_t code;
   int64_t createdTime;
   SRpcMsg rpcMsg;
   int32_t contLen;

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

@@ -27,6 +27,7 @@ void    mndCleanupMnode(SMnode *pMnode);
 bool    mndIsMnode(SMnode *pMnode, int32_t dnodeId);
 void    mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet);
 char   *mndGetRoleStr(int32_t role);
+void    mndUpdateMnodeRole(SMnode *pMnode);
 
 #ifdef __cplusplus
 }

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

@@ -828,9 +828,9 @@ static int32_t mndProcessUseDbMsg(SMnodeMsg *pMsg) {
 static int32_t mndProcessSyncDbMsg(SMnodeMsg *pMsg) {
   SMnode     *pMnode = pMsg->pMnode;
   SSyncDbMsg *pSync = pMsg->rpcMsg.pCont;
-  SDbObj     *pDb = mndAcquireDb(pMnode, pMsg->db);
+  SDbObj     *pDb = mndAcquireDb(pMnode, pSync->db);
   if (pDb == NULL) {
-    mError("db:%s, failed to process sync db msg since %s", pMsg->db, terrstr());
+    mError("db:%s, failed to process sync db msg since %s", pSync->db, terrstr());
     return -1;
   }
 
@@ -841,9 +841,9 @@ static int32_t mndProcessSyncDbMsg(SMnodeMsg *pMsg) {
 static int32_t mndProcessCompactDbMsg(SMnodeMsg *pMsg) {
   SMnode        *pMnode = pMsg->pMnode;
   SCompactDbMsg *pCompact = pMsg->rpcMsg.pCont;
-  SDbObj        *pDb = mndAcquireDb(pMnode, pMsg->db);
+  SDbObj        *pDb = mndAcquireDb(pMnode, pCompact->db);
   if (pDb == NULL) {
-    mError("db:%s, failed to process compact db msg since %s", pMsg->db, terrstr());
+    mError("db:%s, failed to process compact db msg since %s", pCompact->db, terrstr());
     return -1;
   }
 

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

@@ -33,4 +33,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw) {
   return code;
 }
 
-bool mndIsMaster(SMnode *pMnode) { return true; }
\ No newline at end of file
+bool mndIsMaster(SMnode *pMnode) {
+  // pMnode->role = TAOS_SYNC_STATE_LEADER;
+  return true;
+}
\ No newline at end of file

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

@@ -622,10 +622,10 @@ void mndTransHandleActionRsp(SMnodeMsg *pMsg) {
   STransAction *pAction = taosArrayGet(pArray, action);
   if (pAction != NULL) {
     pAction->msgReceived = 1;
-    pAction->errCode = pMsg->code;
+    pAction->errCode = pMsg->rpcMsg.code;
   }
 
-  mDebug("trans:%d, action:%d response is received, code:0x%x", transId, action, pMsg->code);
+  mDebug("trans:%d, action:%d response is received, code:0x%x", transId, action, pMsg->rpcMsg.code);
   mndTransExecute(pMnode, pTrans);
 
 HANDLE_ACTION_RSP_OVER:
@@ -696,7 +696,7 @@ static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pA
   for (int32_t action = 0; action < numOfActions; ++action) {
     STransAction *pAction = taosArrayGet(pArray, action);
     if (pAction == NULL) continue;
-    if (pAction->msgSent) continue;
+    if (pAction->msgReceived && pAction->errCode == 0) continue;
 
     int64_t signature = pTrans->id;
     signature = (signature << 32);
@@ -736,6 +736,7 @@ static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pA
     terrno = errorCode;
     return errorCode;
   } else {
+    mDebug("trans:%d, %d of %d actions executed, code:0x%x", pTrans->id, numOfReceivedMsgs, numOfActions, errorCode);
     return TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 }

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

@@ -178,8 +178,10 @@ static int32_t mndExecSteps(SMnode *pMnode) {
     // (*pMnode->reportProgress)(pStep->name, "start initialize");
 
     if ((*pStep->initFp)(pMnode) != 0) {
+      int32_t code = terrno;
       mError("step:%s exec failed since %s, start to cleanup", pStep->name, terrstr());
       mndCleanupSteps(pMnode, pos);
+      terrno = code;
       return -1;
     } else {
       mDebug("step:%s is initialized", pStep->name);

source/dnode/mnode/sdb/src/sdb.c

@@ -16,6 +16,8 @@
 #define _DEFAULT_SOURCE
 #include "sdbInt.h"
 
+static int32_t sdbCreateDir(SSdb *pSdb);
+
 SSdb *sdbInit(SSdbOpt *pOption) {
   mDebug("start to init sdb in %s", pOption->path);
 
@@ -40,6 +42,11 @@ SSdb *sdbInit(SSdbOpt *pOption) {
     return NULL;
   }
 
+  if (sdbCreateDir(pSdb) != 0) {
+    sdbCleanup(pSdb);
+    return NULL;
+  }
+
   for (ESdbType i = 0; i < SDB_MAX; ++i) {
     taosInitRWLatch(&pSdb->locks[i]);
   }
@@ -53,8 +60,8 @@ void sdbCleanup(SSdb *pSdb) {
   mDebug("start to cleanup sdb");
 
   // if (pSdb->curVer != pSdb->lastCommitVer) {
-    mDebug("write sdb file for curVer:% " PRId64 " and lastVer:%" PRId64, pSdb->curVer, pSdb->lastCommitVer);
-    sdbWriteFile(pSdb);
+  mDebug("write sdb file for curVer:% " PRId64 " and lastVer:%" PRId64, pSdb->curVer, pSdb->lastCommitVer);
+  sdbWriteFile(pSdb);
   // }
 
   if (pSdb->currDir != NULL) {
@@ -133,4 +140,26 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
   mDebug("sdb table:%d is initialized", sdbType);
 
   return 0;
-}
\ No newline at end of file
+}
+
+static int32_t sdbCreateDir(SSdb *pSdb) {
+  if (taosMkDir(pSdb->currDir) != 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    mError("failed to create dir:%s since %s", pSdb->currDir, terrstr());
+    return -1;
+  }
+
+  if (taosMkDir(pSdb->syncDir) != 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    mError("failed to create dir:%s since %s", pSdb->syncDir, terrstr());
+    return -1;
+  }
+
+  if (taosMkDir(pSdb->tmpDir) != 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    mError("failed to create dir:%s since %s", pSdb->tmpDir, terrstr());
+    return -1;
+  }
+
+  return 0;
+}

source/dnode/mnode/sdb/src/sdbFile.c

@@ -17,28 +17,6 @@
 #include "sdbInt.h"
 #include "tchecksum.h"
 
-static int32_t sdbCreateDir(SSdb *pSdb) {
-  if (taosMkDir(pSdb->currDir) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    mError("failed to create dir:%s since %s", pSdb->currDir, terrstr());
-    return -1;
-  }
-
-  if (taosMkDir(pSdb->syncDir) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    mError("failed to create dir:%s since %s", pSdb->syncDir, terrstr());
-    return -1;
-  }
-
-  if (taosMkDir(pSdb->tmpDir) != 0) {
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    mError("failed to create dir:%s since %s", pSdb->tmpDir, terrstr());
-    return -1;
-  }
-
-  return 0;
-}
-
 static int32_t sdbRunDeployFp(SSdb *pSdb) {
   mDebug("start to deploy sdb");
 
@@ -77,7 +55,7 @@ int32_t sdbReadFile(SSdb *pSdb) {
     free(pRaw);
     terrno = TAOS_SYSTEM_ERROR(errno);
     mError("failed to read file:%s since %s", file, terrstr());
-    return -1;
+    return 0;
   }
 
   while (1) {
@@ -225,10 +203,6 @@ int32_t sdbWriteFile(SSdb *pSdb) {
 }
 
 int32_t sdbDeploy(SSdb *pSdb) {
-  if (sdbCreateDir(pSdb) != 0) {
-    return -1;
-  }
-
   if (sdbRunDeployFp(pSdb) != 0) {
     return -1;
   }

source/libs/index/inc/index_tfile.h

@@ -42,7 +42,12 @@ typedef struct TFileHeader {
 
 #define TFILE_HEADER_SIZE (sizeof(TFileHeader))
 #define TFILE_HEADER_NO_FST (TFILE_HEADER_SIZE - sizeof(int32_t))
-//#define TFILE_HADER_PRE_SIZE (sizeof(uint64_t) + sizeof(int32_t) + sizeof(int32_t))
+
+typedef struct TFileValue {
+  char*   colVal;  // null terminated
+  SArray* tableId;
+  int32_t offset;
+} TFileValue;
 
 typedef struct TFileCacheKey {
   uint64_t suid;

source/libs/index/src/index.c

@@ -41,7 +41,7 @@ static pthread_once_t isInit = PTHREAD_ONCE_INIT;
 static void           indexInit();
 
 static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* term, SArray** result);
-static int indexMergeCacheIntoTindex(SIndex* sIdx);
+static int indexFlushCacheToTindex(SIndex* sIdx);
 
 static void indexInterResultsDestroy(SArray* results);
 static int  indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType, SArray* finalResult);
@@ -49,9 +49,7 @@ static int  indexMergeFinalResults(SArray* interResults, EIndexOperatorType oTyp
 int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
   pthread_once(&isInit, indexInit);
   SIndex* sIdx = calloc(1, sizeof(SIndex));
-  if (sIdx == NULL) {
-    return -1;
-  }
+  if (sIdx == NULL) { return -1; }
 
 #ifdef USE_LUCENE
   index_t* index = index_open(path);
@@ -131,9 +129,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
     int32_t colId = fi->colId;
     int32_t version = index->cVersion;
     int     ret = indexCachePut(index->cache, p, colId, version, uid);
-    if (ret != 0) {
-      return ret;
-    }
+    if (ret != 0) { return ret; }
   }
 #endif
 
@@ -221,9 +217,7 @@ void indexOptsDestroy(SIndexOpts* opts){
 
 SIndexMultiTermQuery* indexMultiTermQueryCreate(EIndexOperatorType opera) {
   SIndexMultiTermQuery* p = (SIndexMultiTermQuery*)malloc(sizeof(SIndexMultiTermQuery));
-  if (p == NULL) {
-    return NULL;
-  }
+  if (p == NULL) { return NULL; }
   p->opera = opera;
   p->query = taosArrayInit(4, sizeof(SIndexTermQuery));
   return p;
@@ -250,9 +244,7 @@ SIndexTerm* indexTermCreate(int64_t            suid,
                             const char*        colVal,
                             int32_t            nColVal) {
   SIndexTerm* t = (SIndexTerm*)calloc(1, (sizeof(SIndexTerm)));
-  if (t == NULL) {
-    return NULL;
-  }
+  if (t == NULL) { return NULL; }
 
   t->suid = suid;
   t->operType = oper;
@@ -332,9 +324,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result
   return 0;
 }
 static void indexInterResultsDestroy(SArray* results) {
-  if (results == NULL) {
-    return;
-  }
+  if (results == NULL) { return; }
 
   size_t sz = taosArrayGetSize(results);
   for (size_t i = 0; i < sz; i++) {
@@ -363,10 +353,10 @@ static int indexMergeFinalResults(SArray* interResults, EIndexOperatorType oType
   }
   return 0;
 }
-static int indexMergeCacheIntoTindex(SIndex* sIdx) {
-  if (sIdx == NULL) {
-    return -1;
-  }
+static int indexFlushCacheToTindex(SIndex* sIdx) {
+  if (sIdx == NULL) { return -1; }
+
   indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid);
+
   return 0;
 }

source/libs/index/src/index_cache.c

@@ -151,8 +151,7 @@ int indexCacheSearch(void* cache, SIndexTermQuery* query, int16_t colId, int32_t
   EIndexQueryType qtype = query->qType;
 
   int32_t keyLen = CACHE_KEY_LEN(term);
-
-  char* buf = calloc(1, keyLen);
+  char*   buf = calloc(1, keyLen);
   if (qtype == QUERY_TERM) {
     //
   } else if (qtype == QUERY_PREFIX) {

source/libs/index/src/index_fst_counting_writer.c

@@ -69,9 +69,9 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
     // ugly code, refactor later
     ctx->file.readOnly = readOnly;
     if (readOnly == false) {
-      ctx->file.fd = tfOpenCreateWriteAppend(tmpFile);
+      ctx->file.fd = tfOpenCreateWriteAppend(path);
     } else {
-      ctx->file.fd = tfOpenReadWrite(tmpFile);
+      ctx->file.fd = tfOpenReadWrite(path);
     }
     if (ctx->file.fd < 0) {
       indexError("open file error %d", errno);
@@ -93,6 +93,7 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
 END:
   if (ctx->type == TMemory) { free(ctx->mem.buf); }
   free(ctx);
+  return NULL;
 }
 void writerCtxDestroy(WriterCtx* ctx) {
   if (ctx->type == TMemory) {

source/libs/index/src/index_tfile.c

@@ -25,12 +25,7 @@
 
 #define TF_TABLE_TATOAL_SIZE(sz) (sizeof(sz) + sz * sizeof(uint64_t))
 
-typedef struct TFileValue {
-  char*   colVal;  // null terminated
-  SArray* tableId;
-  int32_t offset;
-} TFileValue;
-
+static int  tfileStrCompare(const void* a, const void* b);
 static int  tfileValueCompare(const void* a, const void* b, const void* param);
 static void tfileSerialTableIdsToBuf(char* buf, SArray* tableIds);
 
@@ -38,17 +33,18 @@ static int tfileWriteHeader(TFileWriter* writer);
 static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset);
 static int tfileWriteData(TFileWriter* write, TFileValue* tval);
 
-static int  tfileReadLoadHeader(TFileReader* reader);
-static int  tfileReadLoadFst(TFileReader* reader);
-static int  tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
-static void tfileReadRef(TFileReader* reader);
-static void tfileReadUnRef(TFileReader* reader);
+static int  tfileReaderLoadHeader(TFileReader* reader);
+static int  tfileReaderLoadFst(TFileReader* reader);
+static int  tfileReaderLoadTableIds(TFileReader* reader, int32_t offset, SArray* result);
+static void tfileReaderRef(TFileReader* reader);
+static void tfileReaderUnRef(TFileReader* reader);
 
 static int  tfileGetFileList(const char* path, SArray* result);
 static int  tfileRmExpireFile(SArray* result);
 static void tfileDestroyFileName(void* elem);
 static int  tfileCompare(const void* a, const void* b);
 static int  tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version);
+static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version);
 static void tfileSerialCacheKey(TFileCacheKey* key, char* buf);
 
 TFileCache* tfileCacheCreate(const char* path) {
@@ -74,23 +70,12 @@ TFileCache* tfileCacheCreate(const char* path) {
 
     WriterCtx* wc = writerCtxCreate(TFile, file, true, 1024 * 64);
     if (wc == NULL) {
-      indexError("failed to open index:  %s", file);
+      indexError("failed to open index:%s", file);
       goto End;
     }
 
     TFileReader* reader = tfileReaderCreate(wc);
-    if (0 != tfileReadLoadHeader(reader)) {
-      tfileReaderDestroy(reader);
-      indexError("failed to load index header, index file: %s", file);
-      goto End;
-    }
-
-    if (0 != tfileReadLoadFst(reader)) {
-      tfileReaderDestroy(reader);
-      indexError("failed to load index fst, index file: %s", file);
-      goto End;
-    }
-    tfileReadRef(reader);
+    tfileReaderRef(reader);
     // loader fst and validate it
     TFileHeader*  header = &reader->header;
     TFileCacheKey key = {.suid = header->suid, .colName = header->colName, .nColName = strlen(header->colName), .colType = header->colType};
@@ -115,7 +100,7 @@ void tfileCacheDestroy(TFileCache* tcache) {
     TFileReader* p = *reader;
     indexInfo("drop table cache suid: %" PRIu64 ", colName: %s, colType: %d", p->header.suid, p->header.colName, p->header.colType);
 
-    tfileReadUnRef(p);
+    tfileReaderUnRef(p);
     reader = taosHashIterate(tcache->tableCache, reader);
   }
   taosHashCleanup(tcache->tableCache);
@@ -127,7 +112,7 @@ TFileReader* tfileCacheGet(TFileCache* tcache, TFileCacheKey* key) {
   tfileSerialCacheKey(key, buf);
 
   TFileReader* reader = taosHashGet(tcache->tableCache, buf, strlen(buf));
-  tfileReadRef(reader);
+  tfileReaderRef(reader);
 
   return reader;
 }
@@ -139,10 +124,10 @@ void tfileCachePut(TFileCache* tcache, TFileCacheKey* key, TFileReader* reader)
   if (*p != NULL) {
     TFileReader* oldReader = *p;
     taosHashRemove(tcache->tableCache, buf, strlen(buf));
-    tfileReadUnRef(oldReader);
+    tfileReaderUnRef(oldReader);
   }
 
-  tfileReadRef(reader);
+  tfileReaderRef(reader);
   taosHashPut(tcache->tableCache, buf, strlen(buf), &reader, sizeof(void*));
   return;
 }
@@ -153,6 +138,19 @@ TFileReader* tfileReaderCreate(WriterCtx* ctx) {
 
   // T_REF_INC(reader);
   reader->ctx = ctx;
+
+  if (0 != tfileReaderLoadHeader(reader)) {
+    tfileReaderDestroy(reader);
+    indexError("failed to load index header, suid: %" PRIu64 ", colName: %s", reader->header.suid, reader->header.colName);
+    return NULL;
+  }
+
+  if (0 != tfileReaderLoadFst(reader)) {
+    tfileReaderDestroy(reader);
+    indexError("failed to load index fst, suid: %" PRIu64 ", colName: %s", reader->header.suid, reader->header.colName);
+    return NULL;
+  }
+
   return reader;
 }
 void tfileReaderDestroy(TFileReader* reader) {
@@ -174,7 +172,7 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* resul
     FstSlice key = fstSliceCreate(term->colVal, term->nColVal);
     if (fstGet(reader->fst, &key, &offset)) {
       indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex", term->suid, term->colName, term->colVal);
-      ret = tfileReadLoadTableIds(reader, offset, result);
+      ret = tfileReaderLoadTableIds(reader, offset, result);
     } else {
       indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName, term->colVal);
     }
@@ -185,7 +183,7 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SArray* resul
   } else {
     // handle later
   }
-  tfileReadUnRef(reader);
+  tfileReaderUnRef(reader);
   return ret;
 }
 
@@ -214,11 +212,18 @@ TFileWriter* tfileWriterCreate(WriterCtx* ctx, TFileHeader* header) {
 
 int tfileWriterPut(TFileWriter* tw, void* data) {
   // sort by coltype and write to tindex
-  __compar_fn_t fn = getComparFunc(tw->header.colType, 0);
+  __compar_fn_t fn;
+
+  int8_t colType = tw->header.colType;
+  if (colType == TSDB_DATA_TYPE_BINARY || colType == TSDB_DATA_TYPE_NCHAR) {
+    fn = tfileStrCompare;
+  } else {
+    fn = getComparFunc(colType, 0);
+  }
   taosArraySortPWithExt((SArray*)(data), tfileValueCompare, &fn);
 
   int32_t bufLimit = 4096, offset = 0;
-  char*   buf = calloc(1, sizeof(bufLimit));
+  char*   buf = calloc(1, sizeof(char) * bufLimit);
   char*   p = buf;
   int32_t sz = taosArrayGetSize((SArray*)data);
   int32_t fstOffset = tw->offset;
@@ -259,6 +264,11 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
   }
   tfree(buf);
 
+  tw->fb = fstBuilderCreate(tw->ctx, 0);
+  if (tw->fb == NULL) {
+    tfileWriterDestroy(tw);
+    return -1;
+  }
   // write fst
   for (size_t i = 0; i < sz; i++) {
     // TODO, fst batch write later
@@ -267,6 +277,9 @@ int tfileWriterPut(TFileWriter* tw, void* data) {
       //
     }
   }
+  fstBuilderFinish(tw->fb);
+  fstBuilderDestroy(tw->fb);
+  tw->fb = NULL;
   return 0;
 }
 void tfileWriterDestroy(TFileWriter* tw) {
@@ -305,6 +318,12 @@ int indexTFilePut(void* tfile, SIndexTerm* term, uint64_t uid) {
   return 0;
 }
 
+static int tfileStrCompare(const void* a, const void* b) {
+  int ret = strcmp((char*)a, (char*)b);
+  if (ret == 0) { return ret; }
+  return ret < 0 ? -1 : 1;
+}
+
 static int tfileValueCompare(const void* a, const void* b, const void* param) {
   __compar_fn_t fn = *(__compar_fn_t*)param;
 
@@ -326,6 +345,7 @@ static int tfileWriteFstOffset(TFileWriter* tw, int32_t offset) {
   int32_t fstOffset = offset + sizeof(tw->header.fstOffset);
   tw->header.fstOffset = fstOffset;
   if (sizeof(fstOffset) != tw->ctx->write(tw->ctx, (char*)&fstOffset, sizeof(fstOffset))) { return -1; }
+  tw->offset += sizeof(fstOffset);
   return 0;
 }
 static int tfileWriteHeader(TFileWriter* writer) {
@@ -355,16 +375,16 @@ static int tfileWriteData(TFileWriter* write, TFileValue* tval) {
   }
   return 0;
 }
-static int tfileReadLoadHeader(TFileReader* reader) {
+static int tfileReaderLoadHeader(TFileReader* reader) {
   // TODO simple tfile header later
   char buf[TFILE_HEADER_SIZE] = {0};
 
-  int64_t nread = reader->ctx->read(reader->ctx, buf, sizeof(buf));
+  int64_t nread = reader->ctx->readFrom(reader->ctx, buf, sizeof(buf), 0);
   assert(nread == sizeof(buf));
   memcpy(&reader->header, buf, sizeof(buf));
   return 0;
 }
-static int tfileReadLoadFst(TFileReader* reader) {
+static int tfileReaderLoadFst(TFileReader* reader) {
   // current load fst into memory, refactor it later
   static int FST_MAX_SIZE = 16 * 1024;
 
@@ -381,9 +401,9 @@ static int tfileReadLoadFst(TFileReader* reader) {
   free(buf);
   fstSliceDestroy(&st);
 
-  return reader->fst == NULL ? 0 : -1;
+  return reader->fst != NULL ? 0 : -1;
 }
-static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* result) {
+static int tfileReaderLoadTableIds(TFileReader* reader, int32_t offset, SArray* result) {
   int32_t    nid;
   WriterCtx* ctx = reader->ctx;
 
@@ -403,12 +423,12 @@ static int tfileReadLoadTableIds(TFileReader* reader, int32_t offset, SArray* re
   free(buf);
   return 0;
 }
-static void tfileReadRef(TFileReader* reader) {
+static void tfileReaderRef(TFileReader* reader) {
   int ref = T_REF_INC(reader);
   UNUSED(ref);
 }
 
-static void tfileReadUnRef(TFileReader* reader) {
+static void tfileReaderUnRef(TFileReader* reader) {
   int ref = T_REF_DEC(reader);
   if (ref == 0) { tfileReaderDestroy(reader); }
 }
@@ -445,6 +465,10 @@ static int tfileCompare(const void* a, const void* b) {
   return strncmp(aName, bName, aLen > bLen ? aLen : bLen);
 }
 // tfile name suid-colId-version.tindex
+static void tfileGenFileName(char* filename, uint64_t suid, int colId, int version) {
+  sprintf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version);
+  return;
+}
 static int tfileParseFileName(const char* filename, uint64_t* suid, int* colId, int* version) {
   if (3 == sscanf(filename, "%" PRIu64 "-%d-%d.tindex", suid, colId, version)) {
     // read suid & colid & version  success

source/libs/parser/inc/parserInt.h

@@ -68,7 +68,7 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pSqlInfo, SQ
  * @param type
  * @return
  */
-int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen);
+int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStmtInfo* pDcl, char* msgBuf, int32_t msgBufLen);
 
 /**
  * Evaluate the numeric and timestamp arithmetic expression in the WHERE clause.

source/libs/parser/src/astValidate.c

@@ -4306,7 +4306,7 @@ int32_t doCheckForCreateTable(SSqlInfo* pInfo, SMsgBuf* pMsgBuf) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen, int32_t* type, char* msgBuf, int32_t msgBufLen) {
+int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStmtInfo* pDcl, char* msgBuf, int32_t msgBufLen) {
   int32_t code = 0;
 
   SMsgBuf m = {.buf = msgBuf, .len = msgBufLen};
@@ -4357,8 +4357,8 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
         }
       }
 
-      *output = buildUserManipulationMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
-      *type = (pInfo->type == TSDB_SQL_CREATE_USER)? TSDB_MSG_TYPE_CREATE_USER:TSDB_MSG_TYPE_ALTER_USER;
+      pDcl->pMsg = (char*)buildUserManipulationMsg(pInfo, &pDcl->msgLen, pCtx->requestId, msgBuf, msgBufLen);
+      pDcl->msgType = (pInfo->type == TSDB_SQL_CREATE_USER)? TSDB_MSG_TYPE_CREATE_USER:TSDB_MSG_TYPE_ALTER_USER;
       break;
     }
 
@@ -4394,21 +4394,21 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
         }
       }
 
-      *output = buildAcctManipulationMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
-      *type = (pInfo->type == TSDB_SQL_CREATE_ACCT)? TSDB_MSG_TYPE_CREATE_ACCT:TSDB_MSG_TYPE_ALTER_ACCT;
+      pDcl->pMsg = (char*)buildAcctManipulationMsg(pInfo, &pDcl->msgLen, pCtx->requestId, msgBuf, msgBufLen);
+      pDcl->msgType = (pInfo->type == TSDB_SQL_CREATE_ACCT)? TSDB_MSG_TYPE_CREATE_ACCT:TSDB_MSG_TYPE_ALTER_ACCT;
       break;
     }
 
     case TSDB_SQL_DROP_ACCT:
     case TSDB_SQL_DROP_USER: {
-      *output = buildDropUserMsg(pInfo, outputLen, pCtx->requestId, msgBuf, msgBufLen);
-      *type = (pInfo->type == TSDB_SQL_DROP_ACCT)? TSDB_MSG_TYPE_DROP_ACCT:TSDB_MSG_TYPE_DROP_USER;
+      pDcl->pMsg = (char*)buildDropUserMsg(pInfo, &pDcl->msgLen, pCtx->requestId, msgBuf, msgBufLen);
+      pDcl->msgType = (pInfo->type == TSDB_SQL_DROP_ACCT)? TSDB_MSG_TYPE_DROP_ACCT:TSDB_MSG_TYPE_DROP_USER;
       break;
     }
     
     case TSDB_SQL_SHOW: {
-      code = setShowInfo(&pInfo->pMiscInfo->showOpt, pCtx, output, outputLen, pMsgBuf);
-      *type = TSDB_MSG_TYPE_SHOW;
+      code = setShowInfo(&pInfo->pMiscInfo->showOpt, pCtx, (void**)&pDcl->pMsg, &pDcl->msgLen, pMsgBuf);
+      pDcl->msgType = TSDB_MSG_TYPE_SHOW;
       break;
     }
 
@@ -4429,9 +4429,9 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
       SUseDbMsg *pUseDbMsg = (SUseDbMsg *) calloc(1, sizeof(SUseDbMsg));
       tNameExtractFullName(&n, pUseDbMsg->db);
 
-      *output    = pUseDbMsg;
-      *outputLen = sizeof(SUseDbMsg);
-      *type      = TSDB_MSG_TYPE_USE_DB;
+      pDcl->pMsg = (char*)pUseDbMsg;
+      pDcl->msgLen = sizeof(SUseDbMsg);
+      pDcl->msgType = TSDB_MSG_TYPE_USE_DB;
       break;
     }
 
@@ -4457,9 +4457,11 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
         return TSDB_CODE_TSC_INVALID_OPERATION;
       }
 
-      *output    = pCreateMsg;
-      *outputLen = sizeof(SCreateDbMsg);
-      *type      = (pInfo->type == TSDB_SQL_CREATE_DB)? TSDB_MSG_TYPE_CREATE_DB:TSDB_MSG_TYPE_ALTER_DB;
+      strncpy(pCreateMsg->db, token.z, token.n);
+
+      pDcl->pMsg = (char*)pCreateMsg;
+      pDcl->msgLen = sizeof(SCreateDbMsg);
+      pDcl->msgType = (pInfo->type == TSDB_SQL_CREATE_DB)? TSDB_MSG_TYPE_CREATE_DB:TSDB_MSG_TYPE_ALTER_DB;
       break;
     }
 
@@ -4481,9 +4483,9 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
       pDropDbMsg->ignoreNotExists = pInfo->pMiscInfo->existsCheck ? 1 : 0;
       assert(code == TSDB_CODE_SUCCESS && name.type == TSDB_DB_NAME_T);
 
-      *type      = TSDB_MSG_TYPE_DROP_DB;
-      *outputLen = sizeof(SDropDbMsg);
-      *output    = pDropDbMsg;
+      pDcl->msgType = TSDB_MSG_TYPE_DROP_DB;
+      pDcl->msgLen = sizeof(SDropDbMsg);
+      pDcl->pMsg = (char*)pDropDbMsg;
       return TSDB_CODE_SUCCESS;
     }
 
@@ -4494,9 +4496,8 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
         if ((code = doCheckForCreateTable(pInfo, pMsgBuf)) != TSDB_CODE_SUCCESS) {
           return code;
         }
-
-        *output = buildCreateTableMsg(pCreateTable, outputLen, pCtx, pMsgBuf);
-        *type = (pCreateTable->type == TSQL_CREATE_TABLE)? TSDB_MSG_TYPE_CREATE_TABLE:TSDB_MSG_TYPE_CREATE_STB;
+        pDcl->pMsg = (char*)buildCreateTableMsg(pCreateTable, &pDcl->msgLen, pCtx, pMsgBuf);
+        pDcl->msgType = (pCreateTable->type == TSQL_CREATE_TABLE)? TSDB_MSG_TYPE_CREATE_TABLE:TSDB_MSG_TYPE_CREATE_STB;
       } else if (pCreateTable->type == TSQL_CREATE_CTABLE) {
         //        if ((code = doCheckForCreateFromStable(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
         //          return code;
@@ -4511,12 +4512,12 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, void**
     }
 
     case TSDB_SQL_DROP_TABLE: {
-      *output = buildDropTableMsg(pInfo, outputLen, pCtx, pMsgBuf);
-      if (output == NULL) {
+      pDcl->pMsg = (char*)buildDropTableMsg(pInfo, &pDcl->msgLen, pCtx, pMsgBuf);
+      if (pDcl->pMsg == NULL) {
         return terrno;
       }
 
-      *type = TSDB_MSG_TYPE_DROP_STB;
+      pDcl->msgType = TSDB_MSG_TYPE_DROP_STB;
       return TSDB_CODE_SUCCESS;
       break;
     }

source/libs/parser/src/insertParser.c

@@ -909,6 +909,7 @@ int32_t parseInsertSql(SParseContext* pContext, SInsertStmtInfo** pInfo) {
   }
 
   *pInfo = context.pOutput;
+  context.pOutput->nodeType = TSDB_SQL_INSERT;
   context.pOutput->schemaAttache = pContext->schemaAttached;
   context.pOutput->payloadType = PAYLOAD_TYPE_KV;
 

source/libs/parser/src/parser.c

@@ -20,7 +20,7 @@
 #include "function.h"
 #include "insertParser.h"
 
-bool qIsInsertSql(const char* pStr, size_t length) {
+bool isInsertSql(const char* pStr, size_t length) {
   int32_t index = 0;
 
   do {
@@ -31,18 +31,28 @@ bool qIsInsertSql(const char* pStr, size_t length) {
   } while (1);
 }
 
-int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCtx, int32_t *type, void** pOutput, int32_t* outputLen, char* msg, int32_t msgLen) {
-  SSqlInfo info = doGenerateAST(pStr);
+bool qIsDclQuery(const SQueryNode* pQuery) {
+  return TSDB_SQL_INSERT != pQuery->type && TSDB_SQL_SELECT != pQuery->type;
+}
+
+int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
+  SSqlInfo info = doGenerateAST(pCxt->pSql);
   if (!info.valid) {
-    strncpy(msg, info.msg, msgLen);
+    strncpy(pCxt->pMsg, info.msg, pCxt->msgLen);
     terrno = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
     return terrno;
   }
 
   if (!isDqlSqlStatement(&info)) {
-    int32_t code = qParserValidateDclSqlNode(&info, pParseCtx, pOutput, outputLen, type, msg, msgLen);
+    SDclStmtInfo* pDcl = calloc(1, sizeof(SQueryStmtInfo));
+    if (NULL == pDcl) {
+      terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; // set correct error code.
+      return terrno;
+    }
+    pDcl->nodeType = info.type;
+    int32_t code = qParserValidateDclSqlNode(&info, &pCxt->ctx, pDcl, pCxt->pMsg, pCxt->msgLen);
     if (code == TSDB_CODE_SUCCESS) {
-      // do nothing
+      *pQuery = (SQueryNode*)pDcl;
     }
   } else {
     SQueryStmtInfo* pQueryInfo = calloc(1, sizeof(SQueryStmtInfo));
@@ -53,9 +63,9 @@ int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCt
 
     struct SCatalog* pCatalog = NULL;
     int32_t code = catalogGetHandle(NULL, &pCatalog);
-    code = qParserValidateSqlNode(pCatalog, &info, pQueryInfo, pParseCtx->requestId, msg, msgLen);
+    code = qParserValidateSqlNode(pCatalog, &info, pQueryInfo, pCxt->ctx.requestId, pCxt->pMsg, pCxt->msgLen);
     if (code == TSDB_CODE_SUCCESS) {
-      *pOutput = pQueryInfo;
+      *pQuery = (SQueryNode*)pQueryInfo;
     }
   }
 
@@ -63,8 +73,12 @@ int32_t qParseQuerySql(const char* pStr, size_t length, SParseBasicCtx* pParseCt
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qParseInsertSql(SParseContext* pContext, SInsertStmtInfo** pInfo) {
-  return parseInsertSql(pContext, pInfo);
+int32_t qParseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
+  if (isInsertSql(pCxt->pSql, pCxt->sqlLen)) {
+    return parseInsertSql(pCxt, (SInsertStmtInfo**)pQuery);
+  } else {
+    return parseQuerySql(pCxt, pQuery);
+  }
 }
 
 int32_t qParserConvertSql(const char* pStr, size_t length, char** pConvertSql) {

source/libs/parser/test/mockCatalog.cpp

@@ -23,20 +23,20 @@
 namespace {
 
 void generateTestT1(MockCatalogService* mcs) {
-  ITableBuilder& builder = mcs->createTableBuilder("root.test", "t1", TSDB_NORMAL_TABLE, 3)
+  ITableBuilder& builder = mcs->createTableBuilder("test", "t1", TSDB_NORMAL_TABLE, 3)
       .setPrecision(TSDB_TIME_PRECISION_MILLI).setVgid(1).addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
       .addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 20);
   builder.done();
 }
 
 void generateTestST1(MockCatalogService* mcs) {
-  ITableBuilder& builder = mcs->createTableBuilder("root.test", "st1", TSDB_SUPER_TABLE, 3, 2)
+  ITableBuilder& builder = mcs->createTableBuilder("test", "st1", TSDB_SUPER_TABLE, 3, 2)
       .setPrecision(TSDB_TIME_PRECISION_MILLI).addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
       .addTag("tag1", TSDB_DATA_TYPE_INT).addTag("tag2", TSDB_DATA_TYPE_BINARY, 20)
       .addColumn("c1", TSDB_DATA_TYPE_INT).addColumn("c2", TSDB_DATA_TYPE_BINARY, 20);
   builder.done();
-  mcs->createSubTable("root.test", "st1", "st1s1", 1);
-  mcs->createSubTable("root.test", "st1", "st1s2", 2);
+  mcs->createSubTable("test", "st1", "st1s1", 1);
+  mcs->createSubTable("test", "st1", "st1s2", 2);
 }
 
 }

source/libs/parser/test/mockCatalogService.cpp

@@ -94,9 +94,9 @@ public:
     return 0;
   }
 
-  int32_t catalogGetTableMeta(const char* pDBName, const char* pTableName, STableMeta** pTableMeta) const {
+  int32_t catalogGetTableMeta(const char* pDbFullName, const char* pTableName, STableMeta** pTableMeta) const {
     std::unique_ptr<STableMeta> table;
-    int32_t code = copyTableSchemaMeta(pDBName, pTableName, &table);
+    int32_t code = copyTableSchemaMeta(toDbname(pDbFullName), pTableName, &table);
     if (TSDB_CODE_SUCCESS != code) {
       return code;
     }
@@ -104,7 +104,7 @@ public:
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t catalogGetTableHashVgroup(const char* pDBName, const char* pTableName, SVgroupInfo* vgInfo) const {
+  int32_t catalogGetTableHashVgroup(const char* pDbFullName, const char* pTableName, SVgroupInfo* vgInfo) const {
     // todo
     return 0;
   }
@@ -195,6 +195,14 @@ private:
   typedef std::map<std::string, std::shared_ptr<MockTableMeta>> TableMetaCache;
   typedef std::map<std::string, TableMetaCache> DbMetaCache;
 
+  std::string toDbname(const std::string& dbFullName) const {
+    std::string::size_type n = dbFullName.find(".");
+    if (n == std::string::npos) {
+      return dbFullName;
+    }
+    return dbFullName.substr(n + 1);
+  }
+
   std::string ttToString(int8_t tableType) const {
     switch (tableType) {
       case TSDB_SUPER_TABLE:

source/libs/parser/test/parserTests.cpp

@@ -714,12 +714,9 @@ TEST(testCase, show_user_Test) {
   SSqlInfo info1 = doGenerateAST(sql1);
   ASSERT_EQ(info1.valid, true);
 
-  void* output = NULL;
-  int32_t type = 0;
-  int32_t len = 0;
-
+  SDclStmtInfo output;
   SParseBasicCtx ct= {.db= "abc", .acctId = 1, .requestId = 1};
-  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, &len, &type, msg, buf.len);
+  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, msg, buf.len);
   ASSERT_EQ(code, 0);
 
   // convert the show command to be the select query
@@ -738,12 +735,9 @@ TEST(testCase, create_user_Test) {
   ASSERT_EQ(info1.valid, true);
   ASSERT_EQ(isDclSqlStatement(&info1), true);
 
-  void* output = NULL;
-  int32_t type = 0;
-  int32_t len = 0;
-
+  SDclStmtInfo output;
   SParseBasicCtx ct= {.db= "abc", .acctId = 1, .requestId = 1};
-  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, &len, &type, msg, buf.len);
+  int32_t code = qParserValidateDclSqlNode(&info1, &ct, &output, msg, buf.len);
   ASSERT_EQ(code, 0);
 
   destroySqlInfo(&info1);

source/libs/planner/src/physicalPlanJson.c

@@ -845,7 +845,7 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
     return TSDB_CODE_FAILED;
   }
   *str = cJSON_Print(json);
-  *len = strlen(*str);
+  *len = strlen(*str) + 1;
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/test/phyPlanTests.cpp

@@ -33,10 +33,6 @@ protected:
   void pushScan(const string& db, const string& table, int32_t scanOp) {
     shared_ptr<MockTableMeta> meta = mockCatalogService->getTableMeta(db, table);
     EXPECT_TRUE(meta);
-// typedef struct SQueryPlanNode {
-//   SArray             *pExpr;        // the query functions or sql aggregations
-//   int32_t             numOfExpr;  // number of result columns, which is also the number of pExprs
-// } SQueryPlanNode;
     unique_ptr<SQueryPlanNode> scan((SQueryPlanNode*)calloc(1, sizeof(SQueryPlanNode)));
     scan->info.type = scanOp;
     scan->numOfCols = meta->schema->tableInfo.numOfColumns;
@@ -54,6 +50,21 @@ protected:
     return code;
   }
 
+  int32_t run(const string& db, const string& sql) {
+    SParseContext cxt;
+    buildParseContext(db, sql, &cxt);
+    SQueryNode* query;
+    int32_t code = qParseQuerySql(&cxt, &query);
+    if (TSDB_CODE_SUCCESS != code) {
+      cout << "error no:" << code << ", msg:" << cxt.pMsg << endl;
+      return code;
+    }
+    SQueryDag* dag = nullptr;
+    code = qCreateQueryDag(query, nullptr, &dag);
+    dag_.reset(dag);
+    return code;
+  }
+
   void explain() {
     size_t level = taosArrayGetSize(dag_->pSubplans);
     for (size_t i = 0; i < level; ++i) {
@@ -64,7 +75,8 @@ protected:
         std::cout << "no " << j << ":" << std::endl;
         int32_t len = 0;
         char* str = nullptr;
-        ASSERT_EQ (TSDB_CODE_SUCCESS, qSubPlanToString((const SSubplan*)taosArrayGetP(subplans, j), &str, &len));
+        ASSERT_EQ(TSDB_CODE_SUCCESS, qSubPlanToString((const SSubplan*)taosArrayGetP(subplans, j), &str, &len));
+        std::cout << "len:" << len << std::endl;
         std::cout << str << std::endl;
         free(str);
       }
@@ -108,6 +120,25 @@ private:
     return info;
   }
 
+  void buildParseContext(const string& db, const string& sql, SParseContext* pCxt) {
+    static string _db;
+    static string _sql;
+    static const int32_t _msgMaxLen = 4096;
+    static char _msg[_msgMaxLen];
+  
+    _db = db;
+    _sql = sql;
+    memset(_msg, 0, _msgMaxLen);
+  
+    pCxt->ctx.acctId = 1;
+    pCxt->ctx.db = _db.c_str();
+    pCxt->ctx.requestId = 1;
+    pCxt->pSql = _sql.c_str();
+    pCxt->sqlLen = _sql.length();
+    pCxt->pMsg = _msg;
+    pCxt->msgLen = _msgMaxLen;
+  }
+
   shared_ptr<MockTableMeta> meta_;
   unique_ptr<SQueryPlanNode> logicPlan_;
   unique_ptr<SQueryDag> dag_;
@@ -115,7 +146,7 @@ private:
 
 // select * from table
 TEST_F(PhyPlanTest, tableScanTest) {
-  pushScan("root.test", "t1", QNODE_TABLESCAN);
+  pushScan("test", "t1", QNODE_TABLESCAN);
   ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
   explain();
   SQueryDag* dag = reslut();
@@ -124,9 +155,17 @@ TEST_F(PhyPlanTest, tableScanTest) {
 
 // select * from supertable
 TEST_F(PhyPlanTest, superTableScanTest) {
-  pushScan("root.test", "st1", QNODE_TABLESCAN);
+  pushScan("test", "st1", QNODE_TABLESCAN);
   ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
   explain();
   SQueryDag* dag = reslut();
   // todo check
 }
+
+// insert into t values(...)
+TEST_F(PhyPlanTest, insertTest) {
+  ASSERT_EQ(run("test", "insert into t1 values (now, 1, \"beijing\")"), TSDB_CODE_SUCCESS);
+  explain();
+  SQueryDag* dag = reslut();
+  // todo check
+}

source/util/src/tworker.c

@@ -38,6 +38,7 @@ int32_t tWorkerInit(SWorkerPool *pool) {
 void tWorkerCleanup(SWorkerPool *pool) {
   for (int i = 0; i < pool->max; ++i) {
     SWorker *worker = pool->workers + i;
+    if (worker == NULL) continue;
     if (taosCheckPthreadValid(worker->thread)) {
       taosQsetThreadResume(pool->qset);
     }
@@ -45,6 +46,7 @@ void tWorkerCleanup(SWorkerPool *pool) {
 
   for (int i = 0; i < pool->max; ++i) {
     SWorker *worker = pool->workers + i;
+    if (worker == NULL) continue;
     if (taosCheckPthreadValid(worker->thread)) {
       pthread_join(worker->thread, NULL);
     }