fix conflicts

example/src/tmq.c

@@ -80,8 +80,9 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  const char* sql = "select * from tu1";
+  /*const char* sql = "select * from tu1";*/
-  pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
+  /*pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));*/
+  pRes = taos_query(pConn, "create topic test_stb_topic_1 as select * from tu1");
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic test_stb_topic_1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/client/taos.h

@@ -26,9 +26,11 @@ extern "C" {
 typedef void   TAOS;
 typedef void   TAOS_STMT;
 typedef void   TAOS_RES;
-typedef void   TAOS_STREAM;
-typedef void   TAOS_SUB;
 typedef void **TAOS_ROW;
+#if 0
+typedef void   TAOS_STREAM;
+#endif
+typedef void   TAOS_SUB;
 
 // Data type definition
 #define TSDB_DATA_TYPE_NULL       0   // 1 bytes
@@ -149,6 +151,7 @@ DLL_EXPORT int       taos_stmt_execute(TAOS_STMT *stmt);
 DLL_EXPORT TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt);
 DLL_EXPORT int       taos_stmt_close(TAOS_STMT *stmt);
 DLL_EXPORT char     *taos_stmt_errstr(TAOS_STMT *stmt);
+DLL_EXPORT int       taos_stmt_affected_rows(TAOS_STMT *stmt);
 
 DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
 DLL_EXPORT TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen);
@@ -179,15 +182,20 @@ DLL_EXPORT int         taos_errno(TAOS_RES *tres);
 DLL_EXPORT void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param);
 DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param);
 
+// Shuduo: temporary enable for app build
+#if 1
 typedef void (*__taos_sub_fn_t)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
 DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char *topic, const char *sql, __taos_sub_fn_t fp,
                                     void *param, int interval);
 DLL_EXPORT TAOS_RES *taos_consume(TAOS_SUB *tsub);
 DLL_EXPORT void      taos_unsubscribe(TAOS_SUB *tsub, int keepProgress);
+#endif
 
+#if 0
 DLL_EXPORT TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sql, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
                                          int64_t stime, void *param, void (*callback)(void *));
 DLL_EXPORT void         taos_close_stream(TAOS_STREAM *tstr);
+#endif
 
 DLL_EXPORT int       taos_load_table_info(TAOS *taos, const char *tableNameList);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
@@ -215,8 +223,12 @@ DLL_EXPORT tmq_list_t *tmq_list_new();
 DLL_EXPORT int32_t     tmq_list_append(tmq_list_t *, const char *);
 DLL_EXPORT void        tmq_list_destroy(tmq_list_t *);
 
+// will be removed in 3.0
 DLL_EXPORT tmq_t *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen);
+
+// will replace last one
 DLL_EXPORT tmq_t *tmq_consumer_new1(tmq_conf_t *conf, char *errstr, int32_t errstrLen);
+
 DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);
 
 /* ------------------------TMQ CONSUMER INTERFACE------------------------ */
@@ -257,18 +269,24 @@ int32_t tmqGetSkipLogNum(tmq_message_t *tmq_message);
 
 DLL_EXPORT TAOS_ROW    tmq_get_row(tmq_message_t *message);
 DLL_EXPORT char       *tmq_get_topic_name(tmq_message_t *message);
-DLL_EXPORT void    *tmq_get_topic_schema(tmq_t *tmq, const char *topic);
+DLL_EXPORT int32_t     tmq_get_vgroup_id(tmq_message_t *message);
+DLL_EXPORT int64_t     tmq_get_request_offset(tmq_message_t *message);
+DLL_EXPORT int64_t     tmq_get_response_offset(tmq_message_t *message);
+DLL_EXPORT TAOS_FIELD *tmq_get_fields(tmq_t *tmq, const char *topic);
+DLL_EXPORT int32_t     tmq_field_count(tmq_t *tmq, const char *topic);
 DLL_EXPORT void        tmq_message_destroy(tmq_message_t *tmq_message);
 
 /* --------------------TMPORARY INTERFACE FOR TESTING--------------------- */
+#if 0
 DLL_EXPORT TAOS_RES *tmq_create_topic(TAOS *taos, const char *name, const char *sql, int sqlLen);
+#endif
 
 DLL_EXPORT TAOS_RES *tmq_create_stream(TAOS *taos, const char *streamName, const char *tbName, const char *sql);
 
-/* -------------------------------- OTHER -------------------------------- */
+/* ------------------------------ TMQ END -------------------------------- */
+#if 1 // Shuduo: temporary enable for app build
 typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
-
+#endif
-DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt);
 
 #ifdef __cplusplus
 }

include/common/tcommon.h

@@ -197,6 +197,11 @@ typedef struct SGroupbyExpr {
   bool    groupbyTag;  // group by tag or column
 } SGroupbyExpr;
 
+enum {
+  FUNC_PARAM_TYPE_VALUE = 0,
+  FUNC_PARAM_TYPE_COLUMN,
+};
+
 typedef struct SFunctParam {
   int32_t  type;
   SColumn* pCol;

include/common/tmsg.h

@@ -2215,23 +2215,6 @@ static FORCE_INLINE void* tDecodeTSmaWrapper(void* buf, STSmaWrapper* pSW) {
   return buf;
 }
 
-typedef struct {
-  int64_t uid;
-  int32_t numOfRows;
-  char*   colData;
-} SMqTbData;
-
-typedef struct {
-  char       topicName[TSDB_TOPIC_FNAME_LEN];
-  int64_t    committedOffset;
-  int64_t    reqOffset;
-  int64_t    rspOffset;
-  int32_t    skipLogNum;
-  int32_t    bodyLen;
-  int32_t    numOfTb;
-  SMqTbData* tbData;
-} SMqTopicData;
-
 typedef struct {
   int8_t  mqMsgType;
   int32_t code;
@@ -2260,7 +2243,10 @@ typedef struct {
 
 typedef struct {
   char        topic[TSDB_TOPIC_FNAME_LEN];
+  int8_t      isSchemaAdaptive;
   SArray*     vgs;  // SArray<SMqSubVgEp>
+  int32_t     numOfFields;
+  TAOS_FIELD* fields;
 } SMqSubTopicEp;
 
 typedef struct {
@@ -2281,32 +2267,6 @@ typedef struct {
   SArray*    topics;  // SArray<SMqSubTopicEp>
 } SMqCMGetSubEpRsp;
 
-typedef struct {
-  int32_t curBlock;
-  int32_t curRow;
-  void**  uData;
-} SMqRowIter;
-
-struct tmq_message_t {
-  SMqPollRsp msg;
-  void*      vg;
-  SMqRowIter iter;
-};
-
-#if 0
-struct tmq_message_t {
-  SMqRspHead head;
-  union {
-    SMqPollRsp       consumeRsp;
-    SMqCMGetSubEpRsp getEpRsp;
-  };
-  void*   extra;
-  int32_t curBlock;
-  int32_t curRow;
-  void**  uData;
-};
-#endif
-
 static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
 
 static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
@@ -2331,17 +2291,21 @@ static FORCE_INLINE void tDeleteSMqCMGetSubEpRsp(SMqCMGetSubEpRsp* pRsp) {
 static FORCE_INLINE int32_t tEncodeSMqSubTopicEp(void** buf, const SMqSubTopicEp* pTopicEp) {
   int32_t tlen = 0;
   tlen += taosEncodeString(buf, pTopicEp->topic);
+  tlen += taosEncodeFixedI8(buf, pTopicEp->isSchemaAdaptive);
   int32_t sz = taosArrayGetSize(pTopicEp->vgs);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
     SMqSubVgEp* pVgEp = (SMqSubVgEp*)taosArrayGet(pTopicEp->vgs, i);
     tlen += tEncodeSMqSubVgEp(buf, pVgEp);
   }
+  tlen += taosEncodeFixedI32(buf, pTopicEp->numOfFields);
+  // tlen += taosEncodeBinary(buf, pTopicEp->fields, pTopicEp->numOfFields * sizeof(TAOS_FIELD));
   return tlen;
 }
 
 static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicEp) {
   buf = taosDecodeStringTo(buf, pTopicEp->topic);
+  buf = taosDecodeFixedI8(buf, &pTopicEp->isSchemaAdaptive);
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
   pTopicEp->vgs = taosArrayInit(sz, sizeof(SMqSubVgEp));
@@ -2353,6 +2317,8 @@ static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicE
     buf = tDecodeSMqSubVgEp(buf, &vgEp);
     taosArrayPush(pTopicEp->vgs, &vgEp);
   }
+  buf = taosDecodeFixedI32(buf, &pTopicEp->numOfFields);
+  // buf = taosDecodeBinary(buf, (void**)&pTopicEp->fields, pTopicEp->numOfFields * sizeof(TAOS_FIELD));
   return buf;
 }
 

include/common/tmsgcb.h

@@ -56,13 +56,14 @@ typedef struct {
   ReleaseHandleFp         releaseHandleFp;
 } SMsgCb;
 
+void    tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb);
 int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq);
 int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype);
 int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq);
 int32_t tmsgSendMnodeReq(const SMsgCb* pMsgCb, SRpcMsg* pReq);
-void    tmsgSendRsp(const SMsgCb* pMsgCb, const SRpcMsg* pRsp);
+void    tmsgSendRsp(const SRpcMsg* pRsp);
 void    tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg);
-void    tmsgReleaseHandle(const SMsgCb* pMsgCb, void* handle, int8_t type);
+void    tmsgReleaseHandle(void* handle, int8_t type);
 
 #ifdef __cplusplus
 }

include/common/ttokendef.h

@@ -139,18 +139,18 @@
 #define TK_TOPIC                          121
 #define TK_AS                             122
 #define TK_NK_BOOL                        123
-#define TK_NK_VARIABLE                    124
+#define TK_NULL                           124
-#define TK_NK_UNDERLINE                   125
+#define TK_NK_VARIABLE                    125
-#define TK_ROWTS                          126
+#define TK_NK_UNDERLINE                   126
-#define TK_TBNAME                         127
+#define TK_ROWTS                          127
-#define TK_QSTARTTS                       128
+#define TK_TBNAME                         128
-#define TK_QENDTS                         129
+#define TK_QSTARTTS                       129
-#define TK_WSTARTTS                       130
+#define TK_QENDTS                         130
-#define TK_WENDTS                         131
+#define TK_WSTARTTS                       131
-#define TK_WDURATION                      132
+#define TK_WENDTS                         132
-#define TK_BETWEEN                        133
+#define TK_WDURATION                      133
-#define TK_IS                             134
+#define TK_BETWEEN                        134
-#define TK_NULL                           135
+#define TK_IS                             135
 #define TK_NK_LT                          136
 #define TK_NK_GT                          137
 #define TK_NK_LE                          138

include/libs/executor/executor.h

@@ -20,8 +20,8 @@
 extern "C" {
 #endif
 
-#include "tcommon.h"
 #include "query.h"
+#include "tcommon.h"
 
 typedef void* qTaskInfo_t;
 typedef void* DataSinkHandle;
@@ -62,7 +62,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type);
  * @param type
  * @return
  */
-int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, void** pBlocks, size_t numOfBlocks, int32_t type);
+int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type);
 
 /**
  * Update the table id list, add or remove.
@@ -167,7 +167,6 @@ int32_t qUpdateQueriedTableIdList(qTaskInfo_t tinfo, int64_t uid, int32_t type);
  */
 void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
 
-
 void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
 
 #ifdef __cplusplus

include/os/os.h

@@ -42,6 +42,9 @@ extern "C" {
 #include <sys/types.h>
 #include <termios.h>
 #include <sys/statvfs.h>
+#include <sys/prctl.h>
+#include <sys/shm.h>
+#include <sys/wait.h>
 
 #if defined(DARWIN)
 #else

include/util/taoserror.h

@@ -16,7 +16,7 @@
 #ifndef _TD_UTIL_TAOS_ERROR_H_
 #define _TD_UTIL_TAOS_ERROR_H_
 
-#include "os.h"
+#include <stdint.h>
 
 #ifdef __cplusplus
 extern "C" {

include/util/tprocess.h

@@ -22,7 +22,7 @@
 extern "C" {
 #endif
 
-typedef enum { PROC_REQ, PROC_RSP, PROC_REGISTER } ProcFuncType;
+typedef enum { PROC_REQ, PROC_RSP, PROC_REG, PROC_RELEASE } ProcFuncType;
 
 typedef struct SProcQueue SProcQueue;
 typedef struct SProcObj   SProcObj;

source/client/src/clientMain.c

@@ -1,12 +1,27 @@
-#include "os.h"
+/*
-#include "tref.h"
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
-#include "trpc.h"
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "catalog.h"
 #include "clientInt.h"
 #include "clientLog.h"
+#include "os.h"
 #include "query.h"
-#include "tmsg.h"
 #include "tglobal.h"
-#include "catalog.h"
+#include "tmsg.h"
+#include "tref.h"
+#include "trpc.h"
 #include "version.h"
 
 #define TSC_VAR_NOT_RELEASE 1
@@ -117,9 +132,7 @@ int  taos_field_count(TAOS_RES *res) {
   return pResInfo->numOfCols;
 }
 
-int  taos_num_fields(TAOS_RES *res) {
+int taos_num_fields(TAOS_RES *res) { return taos_field_count(res); }
-  return taos_field_count(res);
-}
 
 TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
   if (taos_num_fields(res) == 0) {
@@ -144,10 +157,8 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
   }
 
   SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
+  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
-      pRequest->type == TSDB_SQL_INSERT ||
+      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
-      pRequest->code != TSDB_CODE_SUCCESS ||
-      taos_num_fields(res) == 0) {
     return NULL;
   }
 
@@ -249,20 +260,33 @@ int* taos_fetch_lengths(TAOS_RES *res) {
 // todo intergrate with tDataTypes
 const char *taos_data_type(int type) {
   switch (type) {
-    case TSDB_DATA_TYPE_NULL:            return "TSDB_DATA_TYPE_NULL";
+    case TSDB_DATA_TYPE_NULL:
-    case TSDB_DATA_TYPE_BOOL:            return "TSDB_DATA_TYPE_BOOL";
+      return "TSDB_DATA_TYPE_NULL";
-    case TSDB_DATA_TYPE_TINYINT:         return "TSDB_DATA_TYPE_TINYINT";
+    case TSDB_DATA_TYPE_BOOL:
-    case TSDB_DATA_TYPE_SMALLINT:        return "TSDB_DATA_TYPE_SMALLINT";
+      return "TSDB_DATA_TYPE_BOOL";
-    case TSDB_DATA_TYPE_INT:             return "TSDB_DATA_TYPE_INT";
+    case TSDB_DATA_TYPE_TINYINT:
-    case TSDB_DATA_TYPE_BIGINT:          return "TSDB_DATA_TYPE_BIGINT";
+      return "TSDB_DATA_TYPE_TINYINT";
-    case TSDB_DATA_TYPE_FLOAT:           return "TSDB_DATA_TYPE_FLOAT";
+    case TSDB_DATA_TYPE_SMALLINT:
-    case TSDB_DATA_TYPE_DOUBLE:          return "TSDB_DATA_TYPE_DOUBLE";
+      return "TSDB_DATA_TYPE_SMALLINT";
-    case TSDB_DATA_TYPE_VARCHAR:         return "TSDB_DATA_TYPE_VARCHAR";
+    case TSDB_DATA_TYPE_INT:
+      return "TSDB_DATA_TYPE_INT";
+    case TSDB_DATA_TYPE_BIGINT:
+      return "TSDB_DATA_TYPE_BIGINT";
+    case TSDB_DATA_TYPE_FLOAT:
+      return "TSDB_DATA_TYPE_FLOAT";
+    case TSDB_DATA_TYPE_DOUBLE:
+      return "TSDB_DATA_TYPE_DOUBLE";
+    case TSDB_DATA_TYPE_VARCHAR:
+      return "TSDB_DATA_TYPE_VARCHAR";
       //    case TSDB_DATA_TYPE_BINARY:          return "TSDB_DATA_TYPE_VARCHAR";
-    case TSDB_DATA_TYPE_TIMESTAMP:       return "TSDB_DATA_TYPE_TIMESTAMP";
+    case TSDB_DATA_TYPE_TIMESTAMP:
-    case TSDB_DATA_TYPE_NCHAR:           return "TSDB_DATA_TYPE_NCHAR";
+      return "TSDB_DATA_TYPE_TIMESTAMP";
-    case TSDB_DATA_TYPE_JSON:            return "TSDB_DATA_TYPE_JSON";
+    case TSDB_DATA_TYPE_NCHAR:
-    default: return "UNKNOWN";
+      return "TSDB_DATA_TYPE_NCHAR";
+    case TSDB_DATA_TYPE_JSON:
+      return "TSDB_DATA_TYPE_JSON";
+    default:
+      return "UNKNOWN";
   }
 }
 
@@ -335,10 +359,8 @@ int  taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
   }
 
   SRequestObj *pRequest = (SRequestObj *)res;
-  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
+  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
-      pRequest->type == TSDB_SQL_INSERT ||
+      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
-      pRequest->code != TSDB_CODE_SUCCESS ||
-      taos_num_fields(res) == 0) {
     return 0;
   }
 
@@ -352,9 +374,7 @@ int  taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
   return pResultInfo->numOfRows;
 }
 
-int  taos_validate_sql(TAOS *taos, const char *sql) {
+int taos_validate_sql(TAOS *taos, const char *sql) { return true; }
-  return true;
-}
 
 const char *taos_get_server_info(TAOS *taos) {
   if (taos == NULL) {
@@ -387,6 +407,11 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
     // TODO
 }
 
+int taos_load_table_info(TAOS *taos, const char *tableNameList) {
+  // TODO
+  return -1;
+}
+
 TAOS_STMT *taos_stmt_init(TAOS *taos) {
   // TODO
   return NULL;
@@ -437,11 +462,26 @@ int taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name) {
   return -1;
 }
 
+int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
+    // TODO
+    return -1;
+}
+
+int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
+    // TODO
+    return -1;
+}
+
 int taos_stmt_add_batch(TAOS_STMT* stmt) {
     // TODO
     return -1;
 }
 
+TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
+    // TODO
+    return NULL;
+}
+
 int taos_stmt_bind_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind) {
     // TODO
     return -1;

source/client/src/tmq.c

@@ -26,6 +26,18 @@
 #include "tqueue.h"
 #include "tref.h"
 
+typedef struct {
+  int32_t curBlock;
+  int32_t curRow;
+  void**  uData;
+} SMqRowIter;
+
+struct tmq_message_t {
+  SMqPollRsp msg;
+  void*      vg;
+  SMqRowIter iter;
+};
+
 struct tmq_list_t {
   SArray container;
 };
@@ -103,9 +115,10 @@ typedef struct {
   char*       sql;
   char*       topicName;
   int64_t     topicId;
-  int32_t        nextVgIdx;
   SArray*     vgs;  // SArray<SMqClientVg>
-  SSchemaWrapper schema;
+  int8_t      isSchemaAdaptive;
+  int32_t     numOfFields;
+  TAOS_FIELD* fields;
 } SMqClientTopic;
 
 typedef struct {
@@ -131,10 +144,10 @@ typedef struct {
 
 typedef struct {
   tmq_t*         tmq;
-  /*SMqClientVg* pVg;*/
   int32_t        async;
   tsem_t         rspSem;
   tmq_resp_err_t rspErr;
+  /*SMqClientVg* pVg;*/
 } SMqCommitCbParam;
 
 tmq_conf_t* tmq_conf_new() {
@@ -471,7 +484,12 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
     tNameExtractFullName(&name, topicFname);
     tscDebug("subscribe topic: %s", topicFname);
     SMqClientTopic topic = {
-        .nextVgIdx = 0, .sql = NULL, .sqlLen = 0, .topicId = 0, .topicName = topicFname, .vgs = NULL};
+        .sql = NULL,
+        .sqlLen = 0,
+        .topicId = 0,
+        .topicName = topicFname,
+        .vgs = NULL,
+    };
     topic.vgs = taosArrayInit(0, sizeof(SMqClientVg));
     taosArrayPush(tmq->clientTopics, &topic);
     taosArrayPush(req.topicNames, &topicFname);
@@ -615,6 +633,7 @@ _return:
   return pRequest;
 }
 
+#if 0
 TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, int sqlLen) {
   STscObj*     pTscObj = (STscObj*)taos;
   SRequestObj* pRequest = NULL;
@@ -700,6 +719,7 @@ _return:
 
   return pRequest;
 }
+#endif
 
 static char* formatTimestamp(char* buf, int64_t val, int precision) {
   time_t  tt;

source/client/test/tmqTest.cpp

@@ -55,8 +55,9 @@ TEST(testCase, create_topic_ctb_Test) {
 
   taos_free_result(pRes);
 
-  char* sql = "select * from tu";
+  // char* sql = "select * from tu";
-  pRes = tmq_create_topic(pConn, "test_ctb_topic_1", sql, strlen(sql));
+  // pRes = tmq_create_topic(pConn, "test_ctb_topic_1", sql, strlen(sql));
+  pRes = taos_query(pConn, "create test_ctb_topic_1 as select * from tu");
   taos_free_result(pRes);
   taos_close(pConn);
 }
@@ -79,8 +80,9 @@ TEST(testCase, create_topic_stb_Test) {
 
   taos_free_result(pRes);
 
-  char* sql = "select * from st1";
+  // char* sql = "select * from st1";
-  pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
+  // pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
+  pRes = taos_query(pConn, "create test_ctb_topic_1 as select * from st1");
   taos_free_result(pRes);
   taos_close(pConn);
 }

source/common/src/tmsgcb.c

@@ -16,6 +16,10 @@
 #define _DEFAULT_SOURCE
 #include "tmsgcb.h"
 
+static SMsgCb tsDefaultMsgCb;
+
+void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb) { tsDefaultMsgCb = *pMsgCb; }
+
 int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq) {
   return (*pMsgCb->queueFps[qtype])(pMsgCb->pWrapper, pReq);
 }
@@ -32,12 +36,12 @@ int32_t tmsgSendMnodeReq(const SMsgCb* pMsgCb, SRpcMsg* pReq) {
   return (*pMsgCb->sendMnodeReqFp)(pMsgCb->pWrapper, pReq);
 }
 
-void tmsgSendRsp(const SMsgCb* pMsgCb, const SRpcMsg* pRsp) { return (*pMsgCb->sendRspFp)(pMsgCb->pWrapper, pRsp); }
+void tmsgSendRsp(const SRpcMsg* pRsp) { return (*tsDefaultMsgCb.sendRspFp)(tsDefaultMsgCb.pWrapper, pRsp); }
 
 void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg) {
   (*pMsgCb->registerBrokenLinkArgFp)(pMsgCb->pWrapper, pMsg);
 }
 
-void tmsgReleaseHandle(const SMsgCb* pMsgCb, void* handle, int8_t type) {
+void tmsgReleaseHandle(void* handle, int8_t type) {
-  (*pMsgCb->releaseHandleFp)(pMsgCb->pWrapper, handle, type);
+  (*tsDefaultMsgCb.releaseHandleFp)(tsDefaultMsgCb.pWrapper, handle, type);
 }

source/dnode/mgmt/CMakeLists.txt

@@ -1,12 +1,10 @@
-aux_source_directory(src DNODE_SRC)
+aux_source_directory(dm/src DNODE_SRC)
-aux_source_directory(dnode/src DNODE_SRC)
+aux_source_directory(qm/src DNODE_SRC)
-aux_source_directory(qnode/src DNODE_SRC)
+aux_source_directory(bm/src DNODE_SRC)
-aux_source_directory(bnode/src DNODE_SRC)
+aux_source_directory(sm/src DNODE_SRC)
-aux_source_directory(snode/src DNODE_SRC)
+aux_source_directory(vm/src DNODE_SRC)
-aux_source_directory(vnode/src DNODE_SRC)
+aux_source_directory(mm/src DNODE_SRC)
-aux_source_directory(mnode/src DNODE_SRC)
+aux_source_directory(main/src DNODE_SRC)
-aux_source_directory(container/src DNODE_SRC)
-
 add_library(dnode STATIC ${DNODE_SRC})
 target_link_libraries(
     dnode cjson mnode vnode qnode snode bnode wal sync taos tfs monitor
@@ -14,16 +12,22 @@ target_link_libraries(
 target_include_directories(
     dnode
     PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/mgmt"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/dnode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/dm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/qnode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/qm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/bnode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/bm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/snode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/sm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/vnode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/vm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/mnode/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/mm/inc"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/container/inc"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc"
 )
 
-add_subdirectory(main)
+aux_source_directory(main/exe EXEC_SRC)
+add_executable(taosd ${EXEC_SRC})
+target_include_directories(
+    taosd
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main/inc"
+)
+target_link_libraries(taosd dnode)
 
 if(${BUILD_TEST})
     add_subdirectory(test)

source/dnode/mgmt/main/CMakeLists.txt

@@ -1,9 +0,0 @@
-aux_source_directory(src EXEC_SRC)
-add_executable(taosd ${EXEC_SRC})
-
-target_include_directories(
-    taosd
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
-
-target_link_libraries(taosd dnode)

source/dnode/mgmt/main/exe/dndMain.c

@@ -0,0 +1,199 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "dnd.h"
+#include "tconfig.h"
+
+static struct {
+  bool      dumpConfig;
+  bool      generateGrant;
+  bool      printAuth;
+  bool      printVersion;
+  char      envFile[PATH_MAX];
+  char      apolloUrl[PATH_MAX];
+  SArray   *pArgs;  // SConfigPair
+  SDnode   *pDnode;
+  ENodeType ntype;
+} global = {0};
+
+static void dndSigintHandle(int signum, void *info, void *ctx) {
+  dInfo("signal:%d is received", signum);
+  SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
+  if (pDnode != NULL) {
+    dndHandleEvent(pDnode, DND_EVENT_STOP);
+  }
+}
+
+static void dndSetSignalHandle() {
+  taosSetSignal(SIGTERM, dndSigintHandle);
+  taosSetSignal(SIGHUP, dndSigintHandle);
+  taosSetSignal(SIGINT, dndSigintHandle);
+  taosSetSignal(SIGABRT, dndSigintHandle);
+  taosSetSignal(SIGBREAK, dndSigintHandle);
+}
+
+static int32_t dndParseArgs(int32_t argc, char const *argv[]) {
+  for (int32_t i = 1; i < argc; ++i) {
+    if (strcmp(argv[i], "-c") == 0) {
+      if (i < argc - 1) {
+        if (strlen(argv[++i]) >= PATH_MAX) {
+          printf("config file path overflow");
+          return -1;
+        }
+        tstrncpy(configDir, argv[i], PATH_MAX);
+      } else {
+        printf("'-c' requires a parameter, default is %s\n", configDir);
+        return -1;
+      }
+    } else if (strcmp(argv[i], "-a") == 0) {
+      tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
+    } else if (strcmp(argv[i], "-e") == 0) {
+      tstrncpy(global.envFile, argv[++i], PATH_MAX);
+    } else if (strcmp(argv[i], "-k") == 0) {
+      global.generateGrant = true;
+    } else if (strcmp(argv[i], "-n") == 0) {
+      global.ntype = atoi(argv[++i]);
+    } else if (strcmp(argv[i], "-C") == 0) {
+      global.dumpConfig = true;
+    } else if (strcmp(argv[i], "-V") == 0) {
+      global.printVersion = true;
+    } else {
+    }
+  }
+
+  return 0;
+}
+
+static void dndGenerateGrant() {
+  // grantParseParameter();
+  printf("this feature is not implemented yet\n");
+}
+
+static void dndPrintVersion() {
+#ifdef TD_ENTERPRISE
+  char *releaseName = "enterprise";
+#else
+  char *releaseName = "community";
+#endif
+  printf("%s version: %s compatible_version: %s\n", releaseName, version, compatible_version);
+  printf("gitinfo: %s\n", gitinfo);
+  printf("buildInfo: %s\n", buildinfo);
+}
+
+static void dndDumpCfg() {
+  SConfig *pCfg = taosGetCfg();
+  cfgDumpCfg(pCfg, 0, 1);
+}
+
+static SDnodeOpt dndGetOpt() {
+  SConfig  *pCfg = taosGetCfg();
+  SDnodeOpt option = {0};
+
+  option.numOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
+  tstrncpy(option.dataDir, tsDataDir, sizeof(option.dataDir));
+  tstrncpy(option.firstEp, tsFirst, sizeof(option.firstEp));
+  tstrncpy(option.secondEp, tsSecond, sizeof(option.firstEp));
+  option.serverPort = tsServerPort;
+  tstrncpy(option.localFqdn, tsLocalFqdn, sizeof(option.localFqdn));
+  snprintf(option.localEp, sizeof(option.localEp), "%s:%u", option.localFqdn, option.serverPort);
+  option.pDisks = tsDiskCfg;
+  option.numOfDisks = tsDiskCfgNum;
+  return option;
+}
+
+static int32_t dndInitLog() {
+  char logName[12] = {0};
+  snprintf(logName, sizeof(logName), "%slog", dndNodeLogStr(global.ntype));
+  return taosCreateLog(logName, 1, configDir, global.envFile, global.apolloUrl, global.pArgs, 0);
+}
+
+static void dndSetProcName(char **argv) {
+  if (global.ntype != 0) {
+    const char *name = dndNodeProcStr(global.ntype);
+    prctl(PR_SET_NAME, name);
+    strcpy(argv[0], name);
+  }
+}
+
+static int32_t dndRunDnode() {
+  if (dndInit() != 0) {
+    dError("failed to initialize environment since %s", terrstr());
+    return -1;
+  }
+
+  SDnodeOpt option = dndGetOpt();
+  SDnode   *pDnode = dndCreate(&option);
+  if (pDnode == NULL) {
+    dError("failed to to create dnode object since %s", terrstr());
+    return -1;
+  } else {
+    global.pDnode = pDnode;
+    dndSetSignalHandle();
+  }
+
+  dInfo("start the service");
+  int32_t code = dndRun(pDnode);
+  dInfo("start shutting down the service");
+
+  global.pDnode = NULL;
+  dndClose(pDnode);
+  dndCleanup();
+  taosCloseLog();
+  taosCleanupCfg();
+  return code;
+}
+
+int main(int argc, char const *argv[]) {
+  if (!taosCheckSystemIsSmallEnd()) {
+    printf("failed to start since on non-small-end machines\n");
+    return -1;
+  }
+
+  if (dndParseArgs(argc, argv) != 0) {
+    printf("failed to start since parse args error\n");
+    return -1;
+  }
+
+  if (global.generateGrant) {
+    dndGenerateGrant();
+    return 0;
+  }
+
+  if (global.printVersion) {
+    dndPrintVersion();
+    return 0;
+  }
+
+  if (dndInitLog() != 0) {
+    printf("failed to start since init log error\n");
+    return -1;
+  }
+
+  if (taosInitCfg(configDir, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
+    dError("failed to start since read config error");
+    return -1;
+  }
+
+  if (global.dumpConfig) {
+    dndDumpCfg();
+    taosCleanupCfg();
+    taosCloseLog();
+    return 0;
+  }
+
+  dndSetProcName((char **)argv);
+  return dndRunDnode();
+}

source/dnode/mgmt/main/inc/dnd.h

@@ -33,6 +33,7 @@
 #include "tthread.h"
 #include "ttime.h"
 #include "tworker.h"
+#include "tmsgcb.h"
 
 #include "dnode.h"
 #include "monitor.h"
@@ -130,6 +131,8 @@ typedef struct SDnode {
   SMgmtWrapper wrappers[NODE_MAX];
 } SDnode;
 
+const char *dndNodeLogStr(ENodeType ntype);
+const char *dndNodeProcStr(ENodeType ntype);
 EDndStatus  dndGetStatus(SDnode *pDnode);
 void        dndSetStatus(SDnode *pDnode, EDndStatus stat);
 void        dndSetMsgHandle(SMgmtWrapper *pWrapper, int32_t msgType, NodeMsgFp nodeMsgFp, int32_t vgId);
@@ -140,6 +143,7 @@ int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
 int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pMsg);
 void    dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp);
 void    dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg);
+SMsgCb  dndCreateMsgcb(SMgmtWrapper *pWrapper);
 
 int32_t dndProcessNodeMsg(SDnode *pDnode, SNodeMsg *pMsg);
 int32_t dndReadFile(SMgmtWrapper *pWrapper, bool *pDeployed);

source/dnode/mgmt/main/inc/dndMain.h

@@ -1,48 +0,0 @@
-
-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_DND_MAIN_H_
-#define _TD_DND_MAIN_H_
-
-#include "dnode.h"
-
-#include "taoserror.h"
-#include "tconfig.h"
-#include "tglobal.h"
-#include "tlog.h"
-#include "version.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
-#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
-#define dWarn(...)  { if (dDebugFlag & DEBUG_WARN)  { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
-#define dInfo(...)  { if (dDebugFlag & DEBUG_INFO)  { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }}
-#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
-#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
-
-void      dndDumpCfg();
-void      dndPrintVersion();
-void      dndGenerateGrant();
-SDnodeOpt dndGetOpt();
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_DND_MAIN_H_*/

source/dnode/mgmt/main/src/dndExec.c

@@ -77,6 +77,8 @@ static int32_t dndRunInSingleProcess(SDnode *pDnode) {
     SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
     pWrapper->required = dndRequireNode(pWrapper);
     if (!pWrapper->required) continue;
+    SMsgCb msgCb = dndCreateMsgcb(pWrapper);
+    tmsgSetDefaultMsgCb(&msgCb);
 
     if (taosMkDir(pWrapper->path) != 0) {
       terrno = TAOS_SYSTEM_ERROR(errno);
@@ -120,8 +122,7 @@ static void dndConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
                                  ProcFuncType ftype) {
   SRpcMsg *pRpc = &pMsg->rpcMsg;
   pRpc->pCont = pCont;
-  dTrace("msg:%p, get from child queue, type:%s handle:%p app:%p", pMsg, TMSG_INFO(pRpc->msgType), pRpc->handle,
+  dTrace("msg:%p, get from child queue, handle:%p app:%p", pMsg, pRpc->handle, pRpc->ahandle);
-         pRpc->ahandle);
 
   NodeMsgFp msgFp = pWrapper->msgFps[TMSG_INDEX(pRpc->msgType)];
   int32_t   code = (*msgFp)(pWrapper, pMsg);
@@ -142,13 +143,19 @@ static void dndConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
 static void dndConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t msgLen, void *pCont, int32_t contLen,
                                   ProcFuncType ftype) {
   pMsg->pCont = pCont;
-  dTrace("msg:%p, get from parent queue, type:%s handle:%p app:%p", pMsg, TMSG_INFO(pMsg->msgType), pMsg->handle,
+  dTrace("msg:%p, get from parent queue, handle:%p app:%p", pMsg, pMsg->handle, pMsg->ahandle);
-         pMsg->ahandle);
 
   switch (ftype) {
-    case PROC_REGISTER:
+    case PROC_REG:
       rpcRegisterBrokenLinkArg(pMsg);
       break;
+    case PROC_RELEASE:
+      rpcReleaseHandle(pMsg->handle, (int8_t)pMsg->code);
+      rpcFreeCont(pCont);
+      break;
+    case PROC_REQ:
+      // todo send to dnode
+      dndSendReqToMnode(pWrapper, pMsg);
     default:
       dndSendRpcRsp(pWrapper, pMsg);
       break;
@@ -164,6 +171,9 @@ static int32_t dndRunInMultiProcess(SDnode *pDnode) {
     pWrapper->required = dndRequireNode(pWrapper);
     if (!pWrapper->required) continue;
 
+    SMsgCb msgCb = dndCreateMsgcb(pWrapper);
+    tmsgSetDefaultMsgCb(&msgCb);
+
     if (taosMkDir(pWrapper->path) != 0) {
       terrno = TAOS_SYSTEM_ERROR(errno);
       dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
@@ -211,16 +221,24 @@ static int32_t dndRunInMultiProcess(SDnode *pDnode) {
       dndClearNodesExecpt(pDnode, n);
 
       dInfo("node:%s, will be initialized in child process", pWrapper->name);
-      dndOpenNode(pWrapper);
+      if (dndOpenNode(pWrapper) != 0) {
-    } else {
+        dInfo("node:%s, failed to init in child process since %s", pWrapper->name, terrstr());
-      dInfo("node:%s, will not start in parent process, child pid:%d", pWrapper->name, taosProcChildId(pProc));
+        return -1;
-      pWrapper->procType = PROC_PARENT;
       }
 
       if (taosProcRun(pProc) != 0) {
         dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
         return -1;
       }
+      break;
+    } else {
+      dInfo("node:%s, will not start in parent process, child pid:%d", pWrapper->name, taosProcChildId(pProc));
+      pWrapper->procType = PROC_PARENT;
+      if (taosProcRun(pProc) != 0) {
+        dError("node:%s, failed to run proc since %s", pWrapper->name, terrstr());
+        return -1;
+      }
+    }
   }
 
   dndSetStatus(pDnode, DND_STAT_RUNNING);

source/dnode/mgmt/main/src/dndInt.c

@@ -80,19 +80,6 @@ void dndSetStatus(SDnode *pDnode, EDndStatus status) {
   }
 }
 
-const char *dndStatStr(EDndStatus status) {
-  switch (status) {
-    case DND_STAT_INIT:
-      return "init";
-    case DND_STAT_RUNNING:
-      return "running";
-    case DND_STAT_STOPPED:
-      return "stopped";
-    default:
-      return "unknown";
-  }
-}
-
 void dndReportStartup(SDnode *pDnode, const char *pName, const char *pDesc) {
   SStartupReq *pStartup = &pDnode->startup;
   tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);

source/dnode/mgmt/main/src/dndMain.c

@@ -1,138 +0,0 @@
-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "dndMain.h"
-
-static struct {
-  bool    dumpConfig;
-  bool    generateGrant;
-  bool    printAuth;
-  bool    printVersion;
-  char    envFile[PATH_MAX];
-  char    apolloUrl[PATH_MAX];
-  SDnode *pDnode;
-} global = {0};
-
-static void dndSigintHandle(int signum, void *info, void *ctx) {
-  dInfo("signal:%d is received", signum);
-  SDnode *pDnode = atomic_val_compare_exchange_ptr(&global.pDnode, 0, global.pDnode);
-  if (pDnode != NULL) {
-    dndHandleEvent(pDnode, DND_EVENT_STOP);
-  }
-}
-
-static void dndSetSignalHandle() {
-  taosSetSignal(SIGTERM, dndSigintHandle);
-  taosSetSignal(SIGHUP, dndSigintHandle);
-  taosSetSignal(SIGINT, dndSigintHandle);
-  taosSetSignal(SIGABRT, dndSigintHandle);
-  taosSetSignal(SIGBREAK, dndSigintHandle);
-}
-
-static int32_t dndParseOption(int32_t argc, char const *argv[]) {
-  for (int32_t i = 1; i < argc; ++i) {
-    if (strcmp(argv[i], "-c") == 0) {
-      if (i < argc - 1) {
-        if (strlen(argv[++i]) >= PATH_MAX) {
-          printf("config file path overflow");
-          return -1;
-        }
-        tstrncpy(configDir, argv[i], PATH_MAX);
-      } else {
-        printf("'-c' requires a parameter, default is %s\n", configDir);
-        return -1;
-      }
-    } else if (strcmp(argv[i], "-C") == 0) {
-      global.dumpConfig = true;
-    } else if (strcmp(argv[i], "-k") == 0) {
-      global.generateGrant = true;
-    } else if (strcmp(argv[i], "-V") == 0) {
-      global.printVersion = true;
-    } else {
-    }
-  }
-
-  return 0;
-}
-
-static int32_t dndRunDnode() {
-  if (dndInit() != 0) {
-    dInfo("failed to initialize dnode environment since %s", terrstr());
-    return -1;
-  }
-
-  SDnodeOpt option = dndGetOpt();
-
-  SDnode *pDnode = dndCreate(&option);
-  if (pDnode == NULL) {
-    dError("failed to to create dnode object since %s", terrstr());
-    return -1;
-  } else {
-    global.pDnode = pDnode;
-    dndSetSignalHandle();
-  }
-
-  dInfo("start the TDengine service");
-  int32_t code = dndRun(pDnode);
-  dInfo("start shutting down the TDengine service");
-
-  global.pDnode = NULL;
-  dndClose(pDnode);
-  dndCleanup();
-  taosCloseLog();
-  taosCleanupCfg();
-  return code;
-}
-
-int main(int argc, char const *argv[]) {
-  if (!taosCheckSystemIsSmallEnd()) {
-    dError("failed to start TDengine since on non-small-end machines");
-    return -1;
-  }
-
-  if (dndParseOption(argc, argv) != 0) {
-    return -1;
-  }
-
-  if (global.generateGrant) {
-    dndGenerateGrant();
-    return 0;
-  }
-
-  if (global.printVersion) {
-    dndPrintVersion();
-    return 0;
-  }
-
-  if (taosCreateLog("taosdlog", 1, configDir, global.envFile, global.apolloUrl, NULL, 0) != 0) {
-    dError("failed to start TDengine since read log config error");
-    return -1;
-  }
-
-  if (taosInitCfg(configDir, global.envFile, global.apolloUrl, NULL, 0) != 0) {
-    dError("failed to start TDengine since read config error");
-    return -1;
-  }
-
-  if (global.dumpConfig) {
-    dndDumpCfg();
-    taosCleanupCfg();
-    taosCloseLog();
-    return 0;
-  }
-
-  return dndRunDnode();
-}

source/dnode/mgmt/main/src/dndStr.c

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "dndInt.h"
+
+const char *dndStatStr(EDndStatus status) {
+  switch (status) {
+    case DND_STAT_INIT:
+      return "init";
+    case DND_STAT_RUNNING:
+      return "running";
+    case DND_STAT_STOPPED:
+      return "stopped";
+    default:
+      return "unknown";
+  }
+}
+
+const char *dndNodeLogStr(ENodeType ntype) {
+  switch (ntype) {
+    case VNODES:
+      return "vnode";
+    case QNODE:
+      return "qnode";
+    case SNODE:
+      return "snode";
+    case MNODE:
+      return "mnode";
+    case BNODE:
+      return "bnode";
+    default:
+      return "taosd";
+  }
+}
+
+const char *dndNodeProcStr(ENodeType ntype) {
+  switch (ntype) {
+    case VNODES:
+      return "taosv";
+    case QNODE:
+      return "taosq";
+    case SNODE:
+      return "taoss";
+    case MNODE:
+      return "taosm";
+    case BNODE:
+      return "taosb";
+    default:
+      return "taosd";
+  }
+}

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

@@ -320,8 +320,7 @@ static int32_t dndSendRpcReq(STransMgmt *pMgmt, const SEpSet *pEpSet, SRpcMsg *p
 }
 
 int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pReq) {
-  if (pWrapper->procType == PROC_CHILD) {
+  if (pWrapper->procType != PROC_CHILD) {
-  } else {
     SDnode *pDnode = pWrapper->pDnode;
     if (dndGetStatus(pDnode) != DND_STAT_RUNNING) {
       terrno = TSDB_CODE_DND_OFFLINE;
@@ -329,24 +328,25 @@ int32_t dndSendReqToDnode(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg
       return -1;
     }
     return dndSendRpcReq(&pDnode->trans, pEpSet, pReq);
+  } else {
+    while (taosProcPutToParentQ(pWrapper->pProc, pReq, sizeof(SRpcMsg), pReq->pCont, pReq->contLen, PROC_REQ) != 0) {
+      taosMsleep(1);
+    }
   }
 }
 
 int32_t dndSendReqToMnode(SMgmtWrapper *pWrapper, SRpcMsg *pReq) {
-  if (pWrapper->procType == PROC_CHILD) {
-  } else {
   SDnode     *pDnode = pWrapper->pDnode;
   STransMgmt *pTrans = &pDnode->trans;
   SEpSet      epSet = {0};
 
-    SMgmtWrapper *pWrapper = dndAcquireWrapper(pDnode, DNODE);
+  SMgmtWrapper *pWrapper2 = dndAcquireWrapper(pDnode, DNODE);
-    if (pWrapper != NULL) {
+  if (pWrapper2 != NULL) {
-      dmGetMnodeEpSet(pWrapper->pMgmt, &epSet);
+    dmGetMnodeEpSet(pWrapper2->pMgmt, &epSet);
-      dndReleaseWrapper(pWrapper);
+    dndReleaseWrapper(pWrapper2);
   }
   return dndSendRpcReq(pTrans, &epSet, pReq);
 }
-}
 
 void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
   if (pRsp->code == TSDB_CODE_APP_NOT_READY) {
@@ -363,29 +363,44 @@ void dndSendRpcRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
 }
 
 void dndSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
-  if (pWrapper->procType == PROC_CHILD) {
+  if (pWrapper->procType != PROC_CHILD) {
-    int32_t code = -1;
-    do {
-      code = taosProcPutToParentQ(pWrapper->pProc, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_RSP);
-      if (code != 0) {
-        taosMsleep(10);
-      }
-    } while (code != 0);
-  } else {
     dndSendRpcRsp(pWrapper, pRsp);
+  } else {
+    while (taosProcPutToParentQ(pWrapper->pProc, pRsp, sizeof(SRpcMsg), pRsp->pCont, pRsp->contLen, PROC_RSP) != 0) {
+      taosMsleep(1);
+    }
   }
 }
 
 void dndRegisterBrokenLinkArg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
-  if (pWrapper->procType == PROC_CHILD) {
+  if (pWrapper->procType != PROC_CHILD) {
-    int32_t code = -1;
-    do {
-      code = taosProcPutToParentQ(pWrapper->pProc, pMsg, sizeof(SRpcMsg), pMsg->pCont, pMsg->contLen, PROC_REGISTER);
-      if (code != 0) {
-        taosMsleep(10);
-      }
-    } while (code != 0);
-  } else {
     rpcRegisterBrokenLinkArg(pMsg);
+  } else {
+    while (taosProcPutToParentQ(pWrapper->pProc, pMsg, sizeof(SRpcMsg), pMsg->pCont, pMsg->contLen, PROC_REG) != 0) {
+      taosMsleep(1);
     }
   }
+}
+
+static void dndReleaseHandle(SMgmtWrapper *pWrapper, void *handle, int8_t type) {
+  if (pWrapper->procType != PROC_CHILD) {
+    rpcReleaseHandle(handle, type);
+  } else {
+    SRpcMsg msg = {.handle = handle, .code = type};
+    while (taosProcPutToParentQ(pWrapper->pProc, &msg, sizeof(SRpcMsg), NULL, 0, PROC_RELEASE) != 0) {
+      taosMsleep(1);
+    }
+  }
+}
+
+SMsgCb dndCreateMsgcb(SMgmtWrapper *pWrapper) {
+  SMsgCb msgCb = {
+      .pWrapper = pWrapper,
+      .registerBrokenLinkArgFp = dndRegisterBrokenLinkArg,
+      .releaseHandleFp = dndReleaseHandle,
+      .sendMnodeReqFp = dndSendReqToMnode,
+      .sendReqFp = dndSendReqToDnode,
+      .sendRspFp = dndSendRsp,
+  };
+  return msgCb;
+}

source/dnode/mgmt/main/src/dndUtil.c

@@ -1,55 +0,0 @@
-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "dndMain.h"
-
-void dndGenerateGrant() {
-#if 0
-  grantParseParameter();
-#endif
-}
-
-void dndPrintVersion() {
-#ifdef TD_ENTERPRISE
-  char *releaseName = "enterprise";
-#else
-  char *releaseName = "community";
-#endif
-  printf("%s version: %s compatible_version: %s\n", releaseName, version, compatible_version);
-  printf("gitinfo: %s\n", gitinfo);
-  printf("builuInfo: %s\n", buildinfo);
-}
-
-void dndDumpCfg() {
-  SConfig *pCfg = taosGetCfg();
-  cfgDumpCfg(pCfg, 0, 1);
-}
-
-SDnodeOpt dndGetOpt() {
-  SConfig  *pCfg = taosGetCfg();
-  SDnodeOpt option = {0};
-
-  option.numOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
-  tstrncpy(option.dataDir, tsDataDir, sizeof(option.dataDir));
-  tstrncpy(option.firstEp, tsFirst, sizeof(option.firstEp));
-  tstrncpy(option.secondEp, tsSecond, sizeof(option.firstEp));
-  option.serverPort = tsServerPort;
-  tstrncpy(option.localFqdn, tsLocalFqdn, sizeof(option.localFqdn));
-  snprintf(option.localEp, sizeof(option.localEp), "%s:%u", option.localFqdn, option.serverPort);
-  option.pDisks = tsDiskCfg;
-  option.numOfDisks = tsDiskCfgNum;
-  return option;
-}

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

@@ -20,7 +20,8 @@
 #define SYSTABLE_SCH_DB_NAME_LEN    ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
 
 //!!!! Note: only APPEND columns in below tables, NO insert !!!!
-static const SInfosTableSchema dnodesSchema[] = {{.name = "id",             .bytes = 2,   .type = TSDB_DATA_TYPE_SMALLINT},
+static const SInfosTableSchema dnodesSchema[] = {
+    {.name = "id", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
     {.name = "endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "vnodes", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
     {.name = "max_vnodes", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
@@ -28,21 +29,25 @@ static const SInfosTableSchema dnodesSchema[] = {{.name = "id",             .byt
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "note", .bytes = 256 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
 };
-static const SInfosTableSchema mnodesSchema[] = {{.name = "id",             .bytes = 4,   .type = TSDB_DATA_TYPE_INT},
+static const SInfosTableSchema mnodesSchema[] = {
+    {.name = "id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "role", .bytes = 12 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "role_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
 };
-static const SInfosTableSchema modulesSchema[] = {{.name = "id",             .bytes = 4,   .type = TSDB_DATA_TYPE_INT},
+static const SInfosTableSchema modulesSchema[] = {
+    {.name = "id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "endpoint", .bytes = 134, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "module", .bytes = 10, .type = TSDB_DATA_TYPE_BINARY},
 };
-static const SInfosTableSchema qnodesSchema[] = {{.name = "id",             .bytes = 4,   .type = TSDB_DATA_TYPE_INT},
+static const SInfosTableSchema qnodesSchema[] = {
+    {.name = "id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "endpoint", .bytes = 134, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
 };
-static const SInfosTableSchema userDBSchema[] = {{.name = "name",             .bytes = SYSTABLE_SCH_DB_NAME_LEN,   .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userDBSchema[] = {
+    {.name = "name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "vgroups", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
     {.name = "ntables", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
@@ -59,15 +64,18 @@ static const SInfosTableSchema userDBSchema[] = {{.name = "name",             .b
     {.name = "comp", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT},
     {.name = "cachelast", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT},
     {.name = "precision", .bytes = 3 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
-//                                                 {.name = "update",           .bytes = 1,   .type = TSDB_DATA_TYPE_TINYINT},  // disable update
+    //                                                 {.name = "update",           .bytes = 1,   .type =
+    //                                                 TSDB_DATA_TYPE_TINYINT},  // disable update
 };
-static const SInfosTableSchema userFuncSchema[] = {{.name = "name",           .bytes = 32,   .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userFuncSchema[] = {
+    {.name = "name", .bytes = 32, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "ntables", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "precision", .bytes = 2, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "status", .bytes = 10, .type = TSDB_DATA_TYPE_BINARY},
 };
-static const SInfosTableSchema userIdxSchema[] = {{.name = "db_name",          .bytes = 32,   .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userIdxSchema[] = {
+    {.name = "db_name", .bytes = 32, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "table_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "index_database", .bytes = 32, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "index_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
@@ -75,7 +83,8 @@ static const SInfosTableSchema userIdxSchema[] = {{.name = "db_name",          .
     {.name = "index_type", .bytes = 10, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "index_extensions", .bytes = 256, .type = TSDB_DATA_TYPE_BINARY},
 };
-static const SInfosTableSchema userStbsSchema[] = {{.name = "stable_name",     .bytes = SYSTABLE_SCH_TABLE_NAME_LEN,  .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userStbsSchema[] = {
+    {.name = "stable_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "columns", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
@@ -84,7 +93,8 @@ static const SInfosTableSchema userStbsSchema[] = {{.name = "stable_name",     .
     {.name = "last_update", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "table_comment", .bytes = 1024 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_INT},
 };
-static const SInfosTableSchema userStreamsSchema[] = {{.name = "stream_name",  .bytes = SYSTABLE_SCH_DB_NAME_LEN,  .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userStreamsSchema[] = {
+    {.name = "stream_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "user_name", .bytes = 23, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "dest_table", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
@@ -101,7 +111,8 @@ static const SInfosTableSchema userTblsSchema[] = {
     {.name = "ttl", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "table_comment", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
 };
-static const SInfosTableSchema userTblDistSchema[] = {{.name = "db_name",                .bytes = 32,   .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userTblDistSchema[] = {
+    {.name = "db_name", .bytes = 32, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "table_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "distributed_histogram", .bytes = 500, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "min_of_rows", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
@@ -115,12 +126,14 @@ static const SInfosTableSchema userTblDistSchema[] = {{.name = "db_name",
     {.name = "rows_in_mem", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "seek_header_time", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
 };
-static const SInfosTableSchema userUsersSchema[] = {{.name = "name",           .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE,   .type = TSDB_DATA_TYPE_BINARY},
+static const SInfosTableSchema userUsersSchema[] = {
+    {.name = "name", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "privilege", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "account", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
 };
-static const SInfosTableSchema vgroupsSchema[] = {{.name = "vgroup_id",        .bytes = 4,   .type = TSDB_DATA_TYPE_INT},
+static const SInfosTableSchema vgroupsSchema[] = {
+    {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "tables", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "status", .bytes = 12 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
@@ -136,7 +149,34 @@ static const SInfosTableSchema vgroupsSchema[] = {{.name = "vgroup_id",        .
     {.name = "file_size", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
 };
 
-static const SInfosTableMeta infosMeta[] = {{TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema)},
+// TODO put into perf schema
+static const SInfosTableSchema topicSchema[] = {
+    {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
+    {.name = "sql", .bytes = 1024, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "row_len", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+};
+
+static const SInfosTableSchema consumerSchema[] = {
+    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+    {.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+    // ep
+    // up time
+    // topics
+};
+
+static const SInfosTableSchema subscribeSchema[] = {
+    {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+};
+
+static const SInfosTableMeta infosMeta[] = {
+    {TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema)},
     {TSDB_INS_TABLE_MNODES, mnodesSchema, tListLen(mnodesSchema)},
     {TSDB_INS_TABLE_MODULES, modulesSchema, tListLen(modulesSchema)},
     {TSDB_INS_TABLE_QNODES, qnodesSchema, tListLen(qnodesSchema)},
@@ -251,5 +291,3 @@ void mndCleanupInfos(SMnode *pMnode) {
   pMnode->infosMeta = NULL;
 }
 
-
-

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

@@ -62,8 +62,6 @@ static int32_t mndPersistMqSetConnReq(SMnode *pMnode, STrans *pTrans, const SMqT
 
 static int32_t mndPersistRebalanceMsg(SMnode *pMnode, STrans *pTrans, const SMqConsumerEp *pConsumerEp);
 
-static int32_t mndInitUnassignedVg(SMnode *pMnode, const SMqTopicObj *pTopic, SMqSubscribeObj *pSub);
-
 int32_t mndInitSubscribe(SMnode *pMnode) {
   SSdbTable table = {.sdbType = SDB_SUBSCRIBE,
                      .keyType = SDB_KEY_BINARY,
@@ -98,14 +96,6 @@ static SMqSubscribeObj *mndCreateSubscription(SMnode *pMnode, const SMqTopicObj
     return NULL;
   }
 
-#if 0
-  if (mndInitUnassignedVg(pMnode, pTopic, pSub) < 0) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    tDeleteSMqSubscribeObj(pSub);
-    taosMemoryFree(pSub);
-    return NULL;
-  }
-#endif
   // TODO: disable alter subscribed table
   return pSub;
 }
@@ -210,45 +200,6 @@ static int32_t mndPersistCancelConnReq(SMnode *pMnode, STrans *pTrans, const SMq
   return 0;
 }
 
-#if 0
-static int32_t mndProcessResetOffsetReq(SNodeMsg *pMsg) {
-  SMnode             *pMnode = pMsg->pNode;
-  uint8_t            *str = pMsg->rpcMsg.pCont;
-  SMqCMResetOffsetReq req;
-
-  SCoder decoder;
-  tCoderInit(&decoder, TD_LITTLE_ENDIAN, str, pMsg->rpcMsg.contLen, TD_DECODER);
-  tDecodeSMqCMResetOffsetReq(&decoder, &req);
-
-  SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
-  if (pHash == NULL) {
-    return -1;
-  }
-
-  for (int32_t i = 0; i < req.num; i++) {
-    SMqOffset    *pOffset = &req.offsets[i];
-    SMqVgOffsets *pVgOffset = taosHashGet(pHash, &pOffset->vgId, sizeof(int32_t));
-    if (pVgOffset == NULL) {
-      pVgOffset = taosMemoryMalloc(sizeof(SMqVgOffsets));
-      if (pVgOffset == NULL) {
-        return -1;
-      }
-      pVgOffset->offsets = taosArrayInit(0, sizeof(void *));
-      taosArrayPush(pVgOffset->offsets, &pOffset);
-    }
-    taosHashPut(pHash, &pOffset->vgId, sizeof(int32_t), &pVgOffset, sizeof(void *));
-  }
-
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, &pMsg->rpcMsg);
-  if (pTrans == NULL) {
-    mError("mq-reset-offset: failed since %s", terrstr());
-    return -1;
-  }
-
-  return 0;
-}
-#endif
-
 static int32_t mndProcessGetSubEpReq(SNodeMsg *pMsg) {
   SMnode           *pMnode = pMsg->pNode;
   SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont;
@@ -574,251 +525,6 @@ static int32_t mndProcessDoRebalanceMsg(SNodeMsg *pMsg) {
   return 0;
 }
 
-#if 0
-      for (int32_t j = 0; j < consumerNum; j++) {
-        bool            changed = false;
-        bool            unfished = false;
-
-        bool            canUseLeft = imbalanceSolved < imbalanceVg;
-        bool            mustUseLeft = canUseLeft && (imbalanceVg - imbalanceSolved >= consumerNum - j);
-        ASSERT(imbalanceVg - imbalanceSolved <= consumerNum - j);
-
-        int32_t maxVg = vgEachConsumer + canUseLeft;
-        int32_t minVg = vgEachConsumer + mustUseLeft;
-
-        SMqSubConsumer *pSubConsumer = taosArrayGet(pSub->consumers, j);
-        int32_t         vgThisConsumerBeforeRb = taosArrayGetSize(pSubConsumer->vgInfo);
-        int32_t         vgThisConsumerAfterRb;
-        if (vgThisConsumerBeforeRb > maxVg) {
-          vgThisConsumerAfterRb = maxVg; 
-          imbalanceSolved++;
-          changed = true;
-        } else if (vgThisConsumerBeforeRb < minVg) {
-          vgThisConsumerAfterRb = minVg;
-          if (mustUseLeft) imbalanceSolved++;
-          changed = true;
-        } else {
-          vgThisConsumerAfterRb = vgThisConsumerBeforeRb;
-        }
-
-        if (vgThisConsumerBeforeRb > vgThisConsumerAfterRb) {
-          while (taosArrayGetSize(pSubConsumer->vgInfo) > vgThisConsumerAfterRb) {
-            // put into unassigned
-            SMqConsumerEp *pConsumerEp = taosArrayPop(pSubConsumer->vgInfo);
-            ASSERT(pConsumerEp != NULL);
-            ASSERT(pConsumerEp->consumerId == pSubConsumer->consumerId);
-            taosArrayPush(unassignedVgStash, pConsumerEp);
-          }
-
-        } else if (vgThisConsumerBeforeRb < vgThisConsumerAfterRb) {
-          // assign from unassigned
-          while (taosArrayGetSize(pSubConsumer->vgInfo) < vgThisConsumerAfterRb) {
-            // if no unassgined, save j
-            if (taosArrayGetSize(unassignedVgStash) == 0) {
-              taosArrayPush(unassignedConsumerIdx, &j);
-              unfished = true;
-              break;
-            }
-            // assign vg to consumer
-            SMqConsumerEp *pConsumerEp = taosArrayPop(unassignedVgStash);
-            ASSERT(pConsumerEp != NULL);
-            pConsumerEp->oldConsumerId = pConsumerEp->consumerId;
-            pConsumerEp->consumerId = pSubConsumer->consumerId;
-            taosArrayPush(pSubConsumer->vgInfo, pConsumerEp);
-            // build msg and persist into trans
-          }
-        }
-
-        if (changed && !unfished) {
-          SMqConsumerObj *pRebConsumer = mndAcquireConsumer(pMnode, pSubConsumer->consumerId);
-          pRebConsumer->epoch++;
-          if (vgThisConsumerAfterRb != 0) {
-            atomic_store_32(&pRebConsumer->status, MQ_CONSUMER_STATUS__ACTIVE);
-          } else {
-            atomic_store_32(&pRebConsumer->status, MQ_CONSUMER_STATUS__IDLE);
-          }
-          SSdbRaw *pConsumerRaw = mndConsumerActionEncode(pRebConsumer);
-          sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
-          mndTransAppendRedolog(pTrans, pConsumerRaw);
-          mndReleaseConsumer(pMnode, pRebConsumer);
-          // TODO: save history
-        }
-      }
-
-      for (int32_t j = 0; j < taosArrayGetSize(unassignedConsumerIdx); j++) {
-        bool            canUseLeft = imbalanceSolved < imbalanceVg;
-        int32_t         consumerIdx = *(int32_t *)taosArrayGet(unassignedConsumerIdx, j);
-        SMqSubConsumer *pSubConsumer = taosArrayGet(pSub->consumers, consumerIdx);
-        if (canUseLeft) imbalanceSolved++;
-        // must use
-        int32_t vgThisConsumerAfterRb = taosArrayGetSize(pSubConsumer->vgInfo) + canUseLeft;
-        while (taosArrayGetSize(pSubConsumer->vgInfo) < vgEachConsumer + canUseLeft) {
-          // assign vg to consumer
-          SMqConsumerEp *pConsumerEp = taosArrayPop(unassignedVgStash);
-          ASSERT(pConsumerEp != NULL);
-          pConsumerEp->oldConsumerId = pConsumerEp->consumerId;
-          pConsumerEp->consumerId = pSubConsumer->consumerId;
-          taosArrayPush(pSubConsumer->vgInfo, pConsumerEp);
-          // build msg and persist into trans
-        }
-        SMqConsumerObj *pRebConsumer = mndAcquireConsumer(pMnode, pSubConsumer->consumerId);
-        pRebConsumer->epoch++;
-        atomic_store_32(&pRebConsumer->status, MQ_CONSUMER_STATUS__ACTIVE);
-        SSdbRaw *pConsumerRaw = mndConsumerActionEncode(pRebConsumer);
-        sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
-        mndTransAppendRedolog(pTrans, pConsumerRaw);
-        mndReleaseConsumer(pMnode, pRebConsumer);
-        // TODO: save history
-      }
-#endif
-
-#if 0
-    //update consumer status for the subscribption
-    for (int32_t i = 0; i < taosArrayGetSize(pSub->assigned); i++) {
-      SMqConsumerEp *pCEp = taosArrayGet(pSub->assigned, i);
-      int64_t        consumerId = pCEp->consumerId;
-      if (pCEp->status != -1) {
-        int32_t consumerHbStatus = atomic_fetch_add_32(&pCEp->consumerHbStatus, 1);
-        if (consumerHbStatus < MND_SUBSCRIBE_REBALANCE_CNT) {
-          continue;
-        }
-        // put consumer into lostConsumer
-        SMqConsumerEp* lostConsumer = taosArrayPush(pSub->lostConsumer, pCEp);
-        lostConsumer->qmsg = NULL;
-        // put vg into unassigned
-        taosArrayPush(pSub->unassignedVg, pCEp);
-        // remove from assigned
-        // TODO: swap with last one, reduce size and reset i
-        taosArrayRemove(pSub->assigned, i);
-        // remove from available consumer
-        for (int32_t j = 0; j < taosArrayGetSize(pSub->availConsumer); j++) {
-          if (*(int64_t *)taosArrayGet(pSub->availConsumer, i) == pCEp->consumerId) {
-            taosArrayRemove(pSub->availConsumer, j);
-            break;
-          }
-          // TODO: acquire consumer, set status to unavail
-        }
-#if 0
-        SMqConsumerObj* pConsumer = mndAcquireConsumer(pMnode, consumerId);
-        pConsumer->epoch++;
-        printf("current epoch %ld size %ld", pConsumer->epoch, pConsumer->topics->size);
-        SSdbRaw* pRaw = mndConsumerActionEncode(pConsumer);
-        sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-        sdbWriteNotFree(pMnode->pSdb, pRaw);
-        mndReleaseConsumer(pMnode, pConsumer);
-#endif
-      }
-    }
-    // no available consumer, skip rebalance
-    if (taosArrayGetSize(pSub->availConsumer) == 0) {
-      continue;
-    }
-    taosArrayGet(pSub->availConsumer, 0);
-    // rebalance condition1 : have unassigned vg
-    // assign vg to a consumer, trying to find the least assigned one
-    if ((sz = taosArrayGetSize(pSub->unassignedVg)) > 0) {
-      char *topic = NULL;
-      char *cgroup = NULL;
-      mndSplitSubscribeKey(pSub->key, &topic, &cgroup);
-
-      SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
-      STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);
-      for (int32_t i = 0; i < sz; i++) {
-        int64_t        consumerId = *(int64_t *)taosArrayGet(pSub->availConsumer, pSub->nextConsumerIdx);
-        pSub->nextConsumerIdx = (pSub->nextConsumerIdx + 1) % taosArrayGetSize(pSub->availConsumer);
-
-        SMqConsumerEp *pCEp = taosArrayPop(pSub->unassignedVg);
-        pCEp->oldConsumerId = pCEp->consumerId;
-        pCEp->consumerId = consumerId;
-        taosArrayPush(pSub->assigned, pCEp);
-
-        SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
-        pConsumer->epoch++;
-        SSdbRaw* pConsumerRaw = mndConsumerActionEncode(pConsumer);
-        sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
-        sdbWrite(pMnode->pSdb, pConsumerRaw);
-        mndReleaseConsumer(pMnode, pConsumer);
-
-        void* msg;
-        int32_t msgLen;
-        mndBuildRebalanceMsg(&msg, &msgLen, pTopic, pCEp, cgroup, topic);
-
-        // persist msg
-        STransAction action = {0};
-        action.epSet = pCEp->epSet;
-        action.pCont = msg;
-        action.contLen = sizeof(SMsgHead) + msgLen;
-        action.msgType = TDMT_VND_MQ_SET_CONN;
-        mndTransAppendRedoAction(pTrans, &action);
-
-        // persist data
-        SSdbRaw *pRaw = mndSubActionEncode(pSub);
-        sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-        mndTransAppendRedolog(pTrans, pRaw);
-      }
-
-      if (mndTransPrepare(pMnode, pTrans) != 0) {
-        mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
-      }
-      mndReleaseTopic(pMnode, pTopic);
-      mndTransDrop(pTrans);
-      taosMemoryFreeClear(topic);
-      taosMemoryFreeClear(cgroup);
-    }
-    // rebalance condition2 : imbalance assignment
-  }
-  return 0;
-}
-#endif
-
-#if 0
-static int32_t mndInitUnassignedVg(SMnode *pMnode, const SMqTopicObj *pTopic, SMqSubscribeObj *pSub) {
-  SSdb      *pSdb = pMnode->pSdb;
-  SVgObj    *pVgroup = NULL;
-  SQueryPlan *pPlan = qStringToQueryPlan(pTopic->physicalPlan);
-  SArray    *pArray = NULL;
-  SNodeListNode    *inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
-  SSubplan  *plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
-  SArray    *unassignedVg = pSub->unassignedVg;
-
-  void *pIter = NULL;
-  while (1) {
-    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
-    if (pIter == NULL) break;
-    if (pVgroup->dbUid != pTopic->dbUid) {
-      sdbRelease(pSdb, pVgroup);
-      continue;
-    }
-
-    pSub->vgNum++;
-    plan->execNode.nodeId = pVgroup->vgId;
-    plan->execNode.epset = mndGetVgroupEpset(pMnode, pVgroup);
-
-    if (schedulerConvertDagToTaskList(pPlan, &pArray) < 0) {
-      terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
-      mError("unsupport topic: %s, sql: %s", pTopic->name, pTopic->sql);
-      return -1;
-    }
-
-    SMqConsumerEp consumerEp = {0};
-    consumerEp.status = 0;
-    consumerEp.consumerId = -1;
-    STaskInfo *pTaskInfo = taosArrayGet(pArray, 0);
-    consumerEp.epSet = pTaskInfo->addr.epset;
-    consumerEp.vgId = pTaskInfo->addr.nodeId;
-
-    ASSERT(consumerEp.vgId == pVgroup->vgId);
-    consumerEp.qmsg = strdup(pTaskInfo->msg->msg);
-    taosArrayPush(unassignedVg, &consumerEp);
-    // TODO: free taskInfo
-    taosArrayDestroy(pArray);
-  }
-
-  /*qDestroyQueryDag(pDag);*/
-  return 0;
-}
-#endif
-
 static int32_t mndPersistMqSetConnReq(SMnode *pMnode, STrans *pTrans, const SMqTopicObj *pTopic, const char *cgroup,
                                       const SMqConsumerEp *pConsumerEp) {
   ASSERT(pConsumerEp->oldConsumerId == -1);

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

@@ -67,7 +67,9 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
 
   int32_t logicalPlanLen = strlen(pTopic->logicalPlan) + 1;
   int32_t physicalPlanLen = strlen(pTopic->physicalPlan) + 1;
-  int32_t  size = sizeof(SMqTopicObj) + logicalPlanLen + physicalPlanLen + pTopic->sqlLen + MND_TOPIC_RESERVE_SIZE;
+  int32_t swLen = taosEncodeSSchemaWrapper(NULL, &pTopic->schema);
+  int32_t size =
+      sizeof(SMqTopicObj) + logicalPlanLen + physicalPlanLen + pTopic->sqlLen + swLen + MND_TOPIC_RESERVE_SIZE;
   SSdbRaw *pRaw = sdbAllocRaw(SDB_TOPIC, MND_TOPIC_VER_NUMBER, size);
   if (pRaw == NULL) goto TOPIC_ENCODE_OVER;
 
@@ -86,7 +88,6 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
   SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);
   SDB_SET_BINARY(pRaw, dataPos, pTopic->physicalPlan, physicalPlanLen, TOPIC_ENCODE_OVER);
 
-  int32_t swLen = taosEncodeSSchemaWrapper(NULL, &pTopic->schema);
   void *swBuf = taosMemoryMalloc(swLen);
   if (swBuf == NULL) {
     goto TOPIC_ENCODE_OVER;

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

@@ -771,7 +771,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
                       .ahandle = pTrans->rpcAHandle,
                       .pCont = rpcCont,
                       .contLen = pTrans->rpcRspLen};
-    tmsgSendRsp(&pMnode->msgCb, &rspMsg);
+    tmsgSendRsp(&rspMsg);
     pTrans->rpcHandle = NULL;
     pTrans->rpcRsp = NULL;
     pTrans->rpcRspLen = 0;
@@ -898,7 +898,7 @@ static int32_t mndTransSendActionMsg(SMnode *pMnode, STrans *pTrans, SArray *pAr
       pAction->msgReceived = 0;
       pAction->errCode = 0;
     } else {
-      mDebug("trans:%d, action:%d not send since %s", pTrans->id, action, terrstr());
+      mError("trans:%d, action:%d not send since %s", pTrans->id, action, terrstr());
       return -1;
     }
   }
@@ -938,7 +938,7 @@ static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pA
       return errCode;
     }
   } else {
-    mDebug("trans:%d, %d of %d actions executed, code:0x%04x", pTrans->id, numOfReceived, numOfActions, errCode & 0XFFFF);
+    mDebug("trans:%d, %d of %d actions executing", pTrans->id, numOfReceived, numOfActions);
     return TSDB_CODE_MND_ACTION_IN_PROGRESS;
   }
 }

source/dnode/vnode/src/tq/tq.c

@@ -275,7 +275,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
     pMsg->pCont = NULL;
     pMsg->contLen = 0;
     pMsg->code = -1;
-    rpcSendResponse(pMsg);
+    tmsgSendRsp(pMsg);
     return 0;
   }
 
@@ -340,7 +340,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         pMsg->pCont = buf;
         pMsg->contLen = tlen;
         pMsg->code = 0;
-        rpcSendResponse(pMsg);
+        tmsgSendRsp(pMsg);
         return 0;
       }
     } else {
@@ -356,6 +356,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   void*   buf = rpcMallocCont(tlen);
   if (buf == NULL) {
     pMsg->code = -1;
+    ASSERT(0);
     return -1;
   }
   ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
@@ -367,7 +368,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   pMsg->pCont = buf;
   pMsg->contLen = tlen;
   pMsg->code = 0;
-  rpcSendResponse(pMsg);
+  tmsgSendRsp(pMsg);
   /*}*/
 
   return 0;

source/dnode/vnode/src/tq/tqRead.c

@@ -114,10 +114,9 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
   SSchemaWrapper* pSchemaWrapper = pHandle->pSchemaWrapper;
 
   int32_t numOfRows = pHandle->pBlock->numOfRows;
-  int32_t numOfCols = pHandle->pSchema->numOfCols;
+  /*int32_t numOfCols = pHandle->pSchema->numOfCols;*/
   int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
 
-  // TODO: stable case
   if (colNumNeed > pSchemaWrapper->nCols) {
     colNumNeed = pSchemaWrapper->nCols;
   }
@@ -138,58 +137,24 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
       colNeed++;
     } else {
       SColumnInfoData colInfo = {0};
-      int             sz = numOfRows * pColSchema->bytes;
+      /*int             sz = numOfRows * pColSchema->bytes;*/
       colInfo.info.bytes = pColSchema->bytes;
       colInfo.info.colId = pColSchema->colId;
       colInfo.info.type = pColSchema->type;
 
-      colInfo.pData = taosMemoryCalloc(1, sz);
+      if (blockDataEnsureColumnCapacity(&colInfo, numOfRows) < 0) {
-      if (colInfo.pData == NULL) {
+        taosArrayDestroyEx(pArray, (void (*)(void*))tDeleteSSDataBlock);
-        // TODO free
-        taosArrayDestroy(pArray);
         return NULL;
       }
-
-      blockDataEnsureColumnCapacity(&colInfo, numOfRows);
       taosArrayPush(pArray, &colInfo);
       colMeta++;
       colNeed++;
     }
   }
 
-#if 0
-  int j = 0;
-  for (int32_t i = 0; i < colNumNeed; i++) {
-    col_id_t colId = *(col_id_t*)taosArrayGet(pHandle->pColIdList, i);
-    while (j < pSchemaWrapper->nCols && pSchemaWrapper->pSchema[j].colId < colId) {
-      j++;
-    }
-    if (j >= pSchemaWrapper->nCols) {
-      continue;
-    }
-    SSchema*        pColSchema = &pSchemaWrapper->pSchema[j];
-    SColumnInfoData colInfo = {0};
-    int             sz = numOfRows * pColSchema->bytes;
-    colInfo.info.bytes = pColSchema->bytes;
-    colInfo.info.colId = colId;
-    colInfo.info.type = pColSchema->type;
-
-    colInfo.pData = taosMemoryCalloc(1, sz);
-    if (colInfo.pData == NULL) {
-      // TODO free
-      taosArrayDestroy(pArray);
-      return NULL;
-    }
-
-    blockDataEnsureColumnCapacity(&colInfo, numOfRows);
-    taosArrayPush(pArray, &colInfo);
-  }
-#endif
-
   STSRowIter iter = {0};
   tdSTSRowIterInit(&iter, pTschema);
   STSRow* row;
-  // int32_t kvIdx = 0;
   int32_t curRow = 0;
   tInitSubmitBlkIter(pHandle->pBlock, &pHandle->blkIter);
   while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
@@ -202,25 +167,9 @@ SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
       if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
         break;
       }
-      memcpy(POINTER_SHIFT(pColData->pData, curRow * pColData->info.bytes), sVal.val, pColData->info.bytes);
+      // TODO handle null
+      colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL);
     }
-#if 0
-    for (int32_t i = 0; i < colNumNeed; i++) {
-      SColumnInfoData* pColData = taosArrayGet(pArray, i);
-      STColumn*        pCol = schemaColAt(pTschema, i);
-      // TODO
-      if(pCol->colId != pColData->info.colId) {
-        continue;
-      }
-      // void* val = tdGetMemRowDataOfColEx(row, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx);
-      SCellVal sVal = {0};
-      if (!tdSTSRowIterNext(&iter, pCol->colId, pCol->type, &sVal)) {
-        // TODO: reach end
-        break;
-      }
-      memcpy(POINTER_SHIFT(pColData->pData, curRow * pCol->bytes), sVal.val, pCol->bytes);
-    }
-#endif
     curRow++;
   }
   return pArray;

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

@@ -205,8 +205,7 @@ _exit:
   rpcMsg.contLen = rspLen;
   rpcMsg.code = code;
 
-  rpcSendResponse(&rpcMsg);
+  tmsgSendRsp(&rpcMsg);
-
   return TSDB_CODE_SUCCESS;
 }
 
@@ -276,8 +275,7 @@ static int32_t vnodeGetTableList(SVnode *pVnode, SRpcMsg *pMsg) {
       .code = 0,
   };
 
-  rpcSendResponse(&rpcMsg);
+  tmsgSendRsp(&rpcMsg);
-
   taosArrayDestroyEx(pArray, freeItemHelper);
   return 0;
 }

source/libs/executor/src/executor.c

@@ -14,12 +14,12 @@
  */
 
 #include "executor.h"
-#include "tdatablock.h"
 #include "executorimpl.h"
 #include "planner.h"
+#include "tdatablock.h"
 #include "vnode.h"
 
-static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void** input, size_t numOfBlocks, int32_t type, char* id) {
+static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     if (pOperator->numOfDownstream == 0) {
@@ -44,17 +44,18 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void** input, size_t n
     }
 
     if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
-      if (tqReadHandleSetMsg(pInfo->readerHandle, input[0], 0) < 0) {
+      if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) {
         qError("submit msg messed up when initing stream block, %s" PRIx64, id);
         return TSDB_CODE_QRY_APP_ERROR;
       }
     } else {
       for (int32_t i = 0; i < numOfBlocks; ++i) {
-        SSDataBlock* pDataBlock = input[i];
+        SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
 
         SSDataBlock* p = createOneDataBlock(pDataBlock);
         p->info = pDataBlock->info;
 
+        taosArrayClear(p->pDataBlock);
         taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
         taosArrayPush(pInfo->pBlockLists, &p);
       }
@@ -65,10 +66,10 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void** input, size_t n
 }
 
 int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type) {
-  qSetMultiStreamInput(tinfo, (void**) &input, 1, type);
+  return qSetMultiStreamInput(tinfo, input, 1, type);
 }
 
-int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, void** pBlocks, size_t numOfBlocks, int32_t type) {
+int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {
   if (tinfo == NULL) {
     return TSDB_CODE_QRY_APP_ERROR;
   }

source/libs/function/src/builtins.c

@@ -173,7 +173,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .finalizeFunc = NULL
   },
   {
-    .name = "power",
+    .name = "pow",
     .type = FUNCTION_TYPE_POW,
     .classification = FUNC_MGT_SCALAR_FUNC,
     .checkFunc    = stubCheckAndGetResultType,
@@ -409,13 +409,29 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
       // todo
       break;
 
-    case FUNCTION_TYPE_ABS: {
+    case FUNCTION_TYPE_ABS:
+    case FUNCTION_TYPE_CEIL:
+    case FUNCTION_TYPE_FLOOR:
+    case FUNCTION_TYPE_ROUND: {
       SColumnNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
       int32_t paraType = pParam->node.resType.type;
       pFunc->node.resType = (SDataType) { .bytes = tDataTypes[paraType].bytes, .type = paraType };
       break;
     }
 
+    case FUNCTION_TYPE_SIN:
+    case FUNCTION_TYPE_COS:
+    case FUNCTION_TYPE_TAN:
+    case FUNCTION_TYPE_ASIN:
+    case FUNCTION_TYPE_ACOS:
+    case FUNCTION_TYPE_ATAN:
+    case FUNCTION_TYPE_SQRT:
+    case FUNCTION_TYPE_LOG:
+    case FUNCTION_TYPE_POW: {
+      pFunc->node.resType = (SDataType) { .bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE };
+      break;
+    }
+
     default:
       ASSERT(0); // to found the fault ASAP.
   }

source/libs/index/inc/indexCache.h

@@ -1,6 +1,5 @@
 /*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com> *
- *
  * This program is free software: you can use, redistribute, and/or modify
  * it under the terms of the GNU Affero General Public License, version 3
  * or later ("AGPL"), as published by the Free Software Foundation.
@@ -16,7 +15,7 @@
 #define __INDEX_CACHE_H__
 
 #include "indexInt.h"
-#include "index_util.h"
+#include "indexUtil.h"
 #include "tskiplist.h"
 
 // ----------------- key structure in skiplist ---------------------

source/libs/index/inc/indexFst.h

@@ -20,12 +20,12 @@
 extern "C" {
 #endif
 
+#include "indexFstAutomation.h"
+#include "indexFstCountingWriter.h"
+#include "indexFstNode.h"
+#include "indexFstRegistry.h"
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_fst_automation.h"
-#include "index_fst_counting_writer.h"
-#include "index_fst_node.h"
-#include "index_fst_registry.h"
-#include "index_fst_util.h"
 
 #define OUTPUT_PREFIX(a, b) ((a) > (b) ? (b) : (a)
 

source/libs/index/inc/indexFstAutomation.h

@@ -19,8 +19,8 @@
 extern "C" {
 #endif
 
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_fst_util.h"
 
 typedef struct AutomationCtx AutomationCtx;
 

source/libs/index/inc/indexFstNode.h

@@ -20,9 +20,9 @@
 extern "C" {
 #endif
 
+#include "indexFstCountingWriter.h"
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_fst_counting_writer.h"
-#include "index_fst_util.h"
 
 #define FST_BUILDER_NODE_IS_FINAL(bn) (bn->isFinal)
 #define FST_BUILDER_NODE_TRANS_ISEMPTY(bn) (taosArrayGetSize(bn->trans) == 0)

source/libs/index/inc/indexFstRegistry.h

@@ -19,9 +19,9 @@
 extern "C" {
 #endif
 
+#include "indexFstNode.h"
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_fst_node.h"
-#include "index_fst_util.h"
 
 typedef struct FstRegistryCell {
   CompiledAddr    addr;

source/libs/index/inc/indexFstUtil.h

@@ -20,8 +20,8 @@
 extern "C" {
 #endif
 
+#include "indexFstCommon.h"
 #include "indexInt.h"
-#include "index_fst_common.h"
 
 typedef uint64_t FstType;
 typedef uint64_t CompiledAddr;

source/libs/index/inc/indexSparse.h

@@ -12,21 +12,32 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#ifndef TDENGINE_TSCALARFUNCTION_H
+
-#define TDENGINE_TSCALARFUNCTION_H
+#ifndef _TD_INDEX_SPARSE_H_
+#define _TD_INDEX_SPARSE_H_
+
+#include "tarray.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include "function.h"
+typedef struct FstSparseSet {
-#include "scalar.h"
+  SArray *dense;
-
+  SArray *sparse;
-
+  int32_t size;
+} FstSparseSet;
 
+FstSparseSet *sparSetCreate(int32_t sz);
+void          sparSetDestroy(FstSparseSet *s);
+uint32_t      sparSetLen(FstSparseSet *ss);
+uint32_t      sparSetAdd(FstSparseSet *ss, uint32_t ip);
+uint32_t      sparSetGet(FstSparseSet *ss, uint32_t i);
+bool          sparSetContains(FstSparseSet *ss, uint32_t ip);
+void          sparSetClear(FstSparseSet *ss);
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif  // TDENGINE_TSCALARFUNCTION_H
+#endif

source/libs/index/inc/indexTfile.h

@@ -15,11 +15,11 @@
 #ifndef __INDEX_TFILE_H__
 #define __INDEX_TFILE_H__
 
+#include "indexFst.h"
+#include "indexFstCountingWriter.h"
 #include "indexInt.h"
-#include "index_fst.h"
+#include "indexTfile.h"
-#include "index_fst_counting_writer.h"
+#include "indexUtil.h"
-#include "index_tfile.h"
-#include "index_util.h"
 #include "tlockfree.h"
 
 #ifdef __cplusplus

source/libs/index/src/index.c

@@ -14,11 +14,11 @@
  */
 
 #include "index.h"
+#include "indexCache.h"
+#include "indexComm.h"
 #include "indexInt.h"
-#include "index_cache.h"
+#include "indexTfile.h"
-#include "index_comm.h"
+#include "indexUtil.h"
-#include "index_tfile.h"
-#include "index_util.h"
 #include "tdef.h"
 #include "tsched.h"
 

source/libs/index/src/indexCache.c

@@ -13,9 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "index_cache.h"
+#include "indexCache.h"
-#include "index_comm.h"
+#include "indexComm.h"
-#include "index_util.h"
+#include "indexUtil.h"
 #include "tcompare.h"
 #include "tsched.h"
 

source/libs/index/src/indexFst.c

@@ -13,9 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "index_fst.h"
+#include "indexFst.h"
+#include "indexFstAutomation.h"
 #include "indexInt.h"
-#include "index_fst_automation.h"
 #include "tchecksum.h"
 #include "tcoding.h"
 

source/libs/index/src/indexFstAutomation.c

@@ -13,7 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "index_fst_automation.h"
+#include "indexFstAutomation.h"
 
 StartWithStateValue* startWithStateValueCreate(StartWithStateKind kind, ValueType ty, void* val) {
   StartWithStateValue* sv = taosMemoryCalloc(1, sizeof(StartWithStateValue));

source/libs/index/src/indexFstCommon.c

@@ -12,7 +12,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#include "index_fst_common.h"
+#include "indexFstCommon.h"
 
 const uint8_t COMMON_INPUTS[] = {
     84,   // '\x00'
@@ -274,20 +274,23 @@ const uint8_t COMMON_INPUTS[] = {
 };
 
 const char COMMON_INPUTS_INV[] = {
-    't',    'e',    '/',    'o',    'a',    's',    'r',    'i',    'p',    'c',    'n',    'w',    '.',    'h',    'l',    'm',
+    't',    'e',    '/',    'o',    'a',    's',    'r',    'i',    'p',    'c',    'n',    'w',    '.',    'h',
-    '-',    'd',    'u',    '0',    '1',    '2',    'g',    '=',    ':',    'b',    'f',    '3',    'y',    '5',    '&',    '_',
+    'l',    'm',    '-',    'd',    'u',    '0',    '1',    '2',    'g',    '=',    ':',    'b',    'f',    '3',
-    '4',    'v',    '9',    '6',    '7',    '8',    'k',    '%',    '?',    'x',    'C',    'D',    'A',    'S',    'F',    'I',
+    'y',    '5',    '&',    '_',    '4',    'v',    '9',    '6',    '7',    '8',    'k',    '%',    '?',    'x',
-    'B',    'E',    'j',    'P',    'T',    'z',    'R',    'N',    'M',    '+',    'L',    'O',    'q',    'H',    'G',    'W',
+    'C',    'D',    'A',    'S',    'F',    'I',    'B',    'E',    'j',    'P',    'T',    'z',    'R',    'N',
-    'U',    'V',    ',',    'Y',    'K',    'J',    'Z',    'X',    'Q',    ';',    ')',    '(',    '~',    '[',    ']',    '$',
+    'M',    '+',    'L',    'O',    'q',    'H',    'G',    'W',    'U',    'V',    ',',    'Y',    'K',    'J',
-    '!',    '\'',   '*',    '@',    '\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\t',   '\n',   '\x0b',
+    'Z',    'X',    'Q',    ';',    ')',    '(',    '~',    '[',    ']',    '$',    '!',    '\'',   '*',    '@',
-    '\x0c', '\r',   '\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b',
+    '\x00', '\x01', '\x02', '\x03', '\x04', '\x05', '\x06', '\x07', '\x08', '\t',   '\n',   '\x0b', '\x0c', '\r',
-    '\x1c', '\x1d', '\x1e', '\x1f', ' ',    '"',    '#',    '<',    '>',    '\\',   '^',    '`',    '{',    '|',    '}',    '\x7f',
+    '\x0e', '\x0f', '\x10', '\x11', '\x12', '\x13', '\x14', '\x15', '\x16', '\x17', '\x18', '\x19', '\x1a', '\x1b',
-    '\x80', '\x81', '\x82', '\x83', '\x84', '\x85', '\x86', '\x87', '\x88', '\x89', '\x8a', '\x8b', '\x8c', '\x8d', '\x8e', '\x8f',
+    '\x1c', '\x1d', '\x1e', '\x1f', ' ',    '"',    '#',    '<',    '>',    '\\',   '^',    '`',    '{',    '|',
-    '\x90', '\x91', '\x92', '\x93', '\x94', '\x95', '\x96', '\x97', '\x98', '\x99', '\x9a', '\x9b', '\x9c', '\x9d', '\x9e', '\x9f',
+    '}',    '\x7f', '\x80', '\x81', '\x82', '\x83', '\x84', '\x85', '\x86', '\x87', '\x88', '\x89', '\x8a', '\x8b',
-    '\xa0', '\xa1', '\xa2', '\xa3', '\xa4', '\xa5', '\xa6', '\xa7', '\xa8', '\xa9', '\xaa', '\xab', '\xac', '\xad', '\xae', '\xaf',
+    '\x8c', '\x8d', '\x8e', '\x8f', '\x90', '\x91', '\x92', '\x93', '\x94', '\x95', '\x96', '\x97', '\x98', '\x99',
-    '\xb0', '\xb1', '\xb2', '\xb3', '\xb4', '\xb5', '\xb6', '\xb7', '\xb8', '\xb9', '\xba', '\xbb', '\xbc', '\xbd', '\xbe', '\xbf',
+    '\x9a', '\x9b', '\x9c', '\x9d', '\x9e', '\x9f', '\xa0', '\xa1', '\xa2', '\xa3', '\xa4', '\xa5', '\xa6', '\xa7',
-    '\xc0', '\xc1', '\xc2', '\xc3', '\xc4', '\xc5', '\xc6', '\xc7', '\xc8', '\xc9', '\xca', '\xcb', '\xcc', '\xcd', '\xce', '\xcf',
+    '\xa8', '\xa9', '\xaa', '\xab', '\xac', '\xad', '\xae', '\xaf', '\xb0', '\xb1', '\xb2', '\xb3', '\xb4', '\xb5',
-    '\xd0', '\xd1', '\xd2', '\xd3', '\xd4', '\xd5', '\xd6', '\xd7', '\xd8', '\xd9', '\xda', '\xdb', '\xdc', '\xdd', '\xde', '\xdf',
+    '\xb6', '\xb7', '\xb8', '\xb9', '\xba', '\xbb', '\xbc', '\xbd', '\xbe', '\xbf', '\xc0', '\xc1', '\xc2', '\xc3',
-    '\xe0', '\xe1', '\xe2', '\xe3', '\xe4', '\xe5', '\xe6', '\xe7', '\xe8', '\xe9', '\xea', '\xeb', '\xec', '\xed', '\xee', '\xef',
+    '\xc4', '\xc5', '\xc6', '\xc7', '\xc8', '\xc9', '\xca', '\xcb', '\xcc', '\xcd', '\xce', '\xcf', '\xd0', '\xd1',
-    '\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb', '\xfc', '\xfd', '\xfe', '\xff',
+    '\xd2', '\xd3', '\xd4', '\xd5', '\xd6', '\xd7', '\xd8', '\xd9', '\xda', '\xdb', '\xdc', '\xdd', '\xde', '\xdf',
+    '\xe0', '\xe1', '\xe2', '\xe3', '\xe4', '\xe5', '\xe6', '\xe7', '\xe8', '\xe9', '\xea', '\xeb', '\xec', '\xed',
+    '\xee', '\xef', '\xf0', '\xf1', '\xf2', '\xf3', '\xf4', '\xf5', '\xf6', '\xf7', '\xf8', '\xf9', '\xfa', '\xfb',
+    '\xfc', '\xfd', '\xfe', '\xff',
 };

source/libs/index/src/indexFstCountingWriter.c

@@ -13,10 +13,10 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
+#include "indexFstCountingWriter.h"
-#include "index_fst_counting_writer.h"
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_fst_util.h"
+#include "os.h"
 #include "tutil.h"
 
 static int writeCtxDoWrite(WriterCtx* ctx, uint8_t* buf, int len) {
@@ -82,7 +82,9 @@ static int writeCtxDoFlush(WriterCtx* ctx) {
 
 WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int32_t capacity) {
   WriterCtx* ctx = taosMemoryCalloc(1, sizeof(WriterCtx));
-  if (ctx == NULL) { return NULL; }
+  if (ctx == NULL) {
+    return NULL;
+  }
 
   ctx->type = type;
   if (ctx->type == TFile) {
@@ -126,7 +128,9 @@ WriterCtx* writerCtxCreate(WriterType type, const char* path, bool readOnly, int
 
   return ctx;
 END:
-  if (ctx->type == TMemory) { taosMemoryFree(ctx->mem.buf); }
+  if (ctx->type == TMemory) {
+    taosMemoryFree(ctx->mem.buf);
+  }
   taosMemoryFree(ctx);
   return NULL;
 }
@@ -148,14 +152,18 @@ void writerCtxDestroy(WriterCtx* ctx, bool remove) {
       // stat(ctx->file.buf, &fstat);
       // indexError("write file size: %d", (int)(fstat.st_size));
     }
-    if (remove) { unlink(ctx->file.buf); }
+    if (remove) {
+      unlink(ctx->file.buf);
+    }
   }
   taosMemoryFree(ctx);
 }
 
 FstCountingWriter* fstCountingWriterCreate(void* wrt) {
   FstCountingWriter* cw = taosMemoryCalloc(1, sizeof(FstCountingWriter));
-  if (cw == NULL) { return NULL; }
+  if (cw == NULL) {
+    return NULL;
+  }
 
   cw->wrt = wrt;
   //(void *)(writerCtxCreate(TFile, readOnly));
@@ -169,7 +177,9 @@ void fstCountingWriterDestroy(FstCountingWriter* cw) {
 }
 
 int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
-  if (write == NULL) { return 0; }
+  if (write == NULL) {
+    return 0;
+  }
   // update checksum
   // write data to file/socket or mem
   WriterCtx* ctx = write->wrt;
@@ -182,7 +192,9 @@ int fstCountingWriterWrite(FstCountingWriter* write, uint8_t* buf, uint32_t len)
   return len;
 }
 int fstCountingWriterRead(FstCountingWriter* write, uint8_t* buf, uint32_t len) {
-  if (write == NULL) { return 0; }
+  if (write == NULL) {
+    return 0;
+  }
   WriterCtx* ctx = write->wrt;
   int        nRead = ctx->read(ctx, buf, len);
   // assert(nRead == len);

source/libs/index/src/indexFstNode.c

@@ -12,7 +12,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#include "index_fst_node.h"
+#include "indexFstNode.h"
 
 FstBuilderNode* fstBuilderNodeDefault() {
   FstBuilderNode* bn = taosMemoryMalloc(sizeof(FstBuilderNode));
@@ -22,31 +22,45 @@ FstBuilderNode* fstBuilderNodeDefault() {
   return bn;
 }
 void fstBuilderNodeDestroy(FstBuilderNode* node) {
-  if (node == NULL) { return; }
+  if (node == NULL) {
+    return;
+  }
 
   taosArrayDestroy(node->trans);
   taosMemoryFree(node);
 }
 
 bool fstBuilderNodeEqual(FstBuilderNode* n1, FstBuilderNode* n2) {
-  if (n1 == n2) { return true; }
+  if (n1 == n2) {
-  if (n1 == NULL || n2 == NULL) { return false; }
+    return true;
+  }
+  if (n1 == NULL || n2 == NULL) {
+    return false;
+  }
 
-  if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) { return false; }
+  if (n1->isFinal != n2->isFinal || n1->finalOutput != n2->finalOutput) {
+    return false;
+  }
   size_t s1 = n1->trans ? taosArrayGetSize(n1->trans) : 0;
   size_t s2 = n2->trans ? taosArrayGetSize(n2->trans) : 0;
-  if (s1 != s2) { return false; }
+  if (s1 != s2) {
+    return false;
+  }
   for (size_t i = 0; i < s1; i++) {
     FstTransition* t1 = taosArrayGet(n1->trans, i);
     FstTransition* t2 = taosArrayGet(n2->trans, i);
-    if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) { return false; }
+    if (t1->inp != t2->inp || t1->out != t2->out || t1->addr != t2->addr) {
+      return false;
+    }
   }
 
   return true;
 }
 FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src) {
   FstBuilderNode* node = taosMemoryMalloc(sizeof(FstBuilderNode));
-  if (node == NULL) { return NULL; }
+  if (node == NULL) {
+    return NULL;
+  }
 
   //
   size_t  sz = taosArrayGetSize(src->trans);
@@ -64,7 +78,9 @@ FstBuilderNode* fstBuilderNodeClone(FstBuilderNode* src) {
 }
 // not destroy src, User's bussiness
 void fstBuilderNodeCloneFrom(FstBuilderNode* dst, FstBuilderNode* src) {
-  if (dst == NULL || src == NULL) { return; }
+  if (dst == NULL || src == NULL) {
+    return;
+  }
 
   dst->isFinal = src->isFinal;
   dst->finalOutput = src->finalOutput;

source/libs/index/src/indexFstRegister.c

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "indexFstRegistry.h"
 #include "os.h"
-#include "index_fst_registry.h"
 
 uint64_t fstRegistryHash(FstRegistry* registry, FstBuilderNode* bNode) {
   // TODO(yihaoDeng): refactor later
@@ -35,7 +35,9 @@ uint64_t fstRegistryHash(FstRegistry* registry, FstBuilderNode* bNode) {
 }
 static void fstRegistryCellSwap(SArray* arr, uint32_t a, uint32_t b) {
   size_t sz = taosArrayGetSize(arr);
-  if (a >= sz || b >= sz) { return; }
+  if (a >= sz || b >= sz) {
+    return;
+  }
 
   FstRegistryCell* cell1 = (FstRegistryCell*)taosArrayGet(arr, a);
   FstRegistryCell* cell2 = (FstRegistryCell*)taosArrayGet(arr, b);
@@ -52,7 +54,9 @@ static void fstRegistryCellSwap(SArray* arr, uint32_t a, uint32_t b) {
 
 static void fstRegistryCellPromote(SArray* arr, uint32_t start, uint32_t end) {
   size_t sz = taosArrayGetSize(arr);
-  if (start >= sz && end >= sz) { return; }
+  if (start >= sz && end >= sz) {
+    return;
+  }
 
   assert(start >= end);
 
@@ -66,7 +70,9 @@ static void fstRegistryCellPromote(SArray* arr, uint32_t start, uint32_t end) {
 
 FstRegistry* fstRegistryCreate(uint64_t tableSize, uint64_t mruSize) {
   FstRegistry* registry = taosMemoryMalloc(sizeof(FstRegistry));
-  if (registry == NULL) { return NULL; }
+  if (registry == NULL) {
+    return NULL;
+  }
 
   uint64_t nCells = tableSize * mruSize;
   SArray*  tb = (SArray*)taosArrayInit(nCells, sizeof(FstRegistryCell));
@@ -87,7 +93,9 @@ FstRegistry* fstRegistryCreate(uint64_t tableSize, uint64_t mruSize) {
 }
 
 void fstRegistryDestroy(FstRegistry* registry) {
-  if (registry == NULL) { return; }
+  if (registry == NULL) {
+    return;
+  }
 
   SArray* tb = registry->table;
   size_t  sz = taosArrayGetSize(tb);
@@ -100,7 +108,9 @@ void fstRegistryDestroy(FstRegistry* registry) {
 }
 
 FstRegistryEntry* fstRegistryGetEntry(FstRegistry* registry, FstBuilderNode* bNode) {
-  if (taosArrayGetSize(registry->table) <= 0) { return NULL; }
+  if (taosArrayGetSize(registry->table) <= 0) {
+    return NULL;
+  }
   uint64_t bucket = fstRegistryHash(registry, bNode);
   uint64_t start = registry->mruSize * bucket;
   uint64_t end = start + registry->mruSize;
@@ -165,6 +175,4 @@ FstRegistryEntry* fstRegistryGetEntry(FstRegistry* registry, FstBuilderNode* bNo
   }
   return entry;
 }
-void fstRegistryEntryDestroy(FstRegistryEntry* entry) {
+void fstRegistryEntryDestroy(FstRegistryEntry* entry) { taosMemoryFree(entry); }
-  taosMemoryFree(entry);
-}

source/libs/index/src/indexFstUtil.c

@@ -12,8 +12,8 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#include "index_fst_util.h"
+#include "indexFstUtil.h"
-#include "index_fst_common.h"
+#include "indexFstCommon.h"
 
 // A sentinel value used to indicate an empty final state
 const CompiledAddr EMPTY_ADDRESS = 0;

source/libs/index/src/indexSparse.c

@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "indexSparse.h"
+
+FstSparseSet *sparSetCreate(int32_t sz) {
+  FstSparseSet *ss = taosMemoryCalloc(1, sizeof(FstSparseSet));
+  if (ss = NULL) {
+    return NULL;
+  }
+
+  ss->dense = taosArrayInit(sz, sizeof(uint32_t));
+  ss->sparse = taosArrayInit(sz, sizeof(uint32_t));
+  ss->size = sz;
+  return ss;
+}
+void sparSetDestroy(FstSparseSet *ss) {
+  if (ss == NULL) {
+    return;
+  }
+  taosArrayDestroy(ss->dense);
+  taosArrayDestroy(ss->sparse);
+  taosMemoryFree(ss);
+}
+uint32_t sparSetLen(FstSparseSet *ss) { return ss == NULL ? 0 : ss->size; }
+uint32_t sparSetAdd(FstSparseSet *ss, uint32_t ip) {
+  if (ss == NULL) {
+    return 0;
+  }
+  uint32_t i = ss->size;
+  taosArraySet(ss->dense, i, &ip);
+  taosArraySet(ss->sparse, ip, &i);
+  ss->size += 1;
+  return i;
+}
+uint32_t sparSetGet(FstSparseSet *ss, uint32_t i) {
+  if (i >= taosArrayGetSize(ss->dense)) {
+    return 0;
+  }
+  uint32_t *v = taosArrayGet(ss->dense, i);
+  return *v;
+}
+bool sparSetContains(FstSparseSet *ss, uint32_t ip) {
+  if (ip >= taosArrayGetSize(ss->sparse)) {
+    return false;
+  }
+  uint32_t i = *(uint32_t *)taosArrayGet(ss->sparse, ip);
+  if (i >= taosArrayGetSize(ss->dense)) {
+    return false;
+  }
+  uint32_t v = *(uint32_t *)taosArrayGet(ss->dense, i);
+  return v == ip;
+}
+void sparSetClear(FstSparseSet *ss) {
+  if (ss == NULL) {
+    return;
+  }
+  ss->size = 0;
+}

source/libs/index/src/indexTfile.c

@@ -13,12 +13,12 @@ p *
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "index_tfile.h"
+#include "indexTfile.h"
 #include "index.h"
-#include "index_comm.h"
+#include "indexComm.h"
-#include "index_fst.h"
+#include "indexFst.h"
-#include "index_fst_counting_writer.h"
+#include "indexFstCountingWriter.h"
-#include "index_util.h"
+#include "indexUtil.h"
 #include "taosdef.h"
 #include "tcoding.h"
 #include "tcompare.h"

source/libs/index/src/indexUtil.c

@@ -12,7 +12,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
-#include "index_util.h"
+#include "indexUtil.h"
 #include "index.h"
 #include "tcompare.h"
 

source/libs/index/test/fstTest.cc

@@ -5,12 +5,12 @@
 #include <thread>
 #include <vector>
 #include "index.h"
+#include "indexCache.h"
+#include "indexFst.h"
+#include "indexFstCountingWriter.h"
+#include "indexFstUtil.h"
 #include "indexInt.h"
-#include "index_cache.h"
+#include "indexTfile.h"
-#include "index_fst.h"
-#include "index_fst_counting_writer.h"
-#include "index_fst_util.h"
-#include "index_tfile.h"
 #include "tskiplist.h"
 #include "tutil.h"
 void* callback(void* s) { return s; }