Merge branch 'develop' into todevelop

CMakeLists.txt

@@ -13,6 +13,9 @@ ENDIF ()
 SET(TD_ACCOUNT FALSE)
 SET(TD_ADMIN FALSE)
 SET(TD_GRANT FALSE)
+SET(TD_SYNC TRUE)
+SET(TD_MQTT TRUE)
+SET(TD_TSDB_PLUGINS FALSE)
 
 SET(TD_COVER FALSE)
 SET(TD_MEM_CHECK FALSE)

README.md

@@ -3,6 +3,7 @@
 [![Coverage Status](https://coveralls.io/repos/github/taosdata/TDengine/badge.svg?branch=develop)](https://coveralls.io/github/taosdata/TDengine?branch=develop)
 [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/4201/badge)](https://bestpractices.coreinfrastructure.org/projects/4201)
 [![Docker Pulls](https://img.shields.io/docker/pulls/tdengine/tdengine)](https://hub.docker.com/repository/docker/tdengine/tdengine)
+[![tdengine](https://snapcraft.io//tdengine/badge.svg)](https://snapcraft.io/tdengine)
 
 [![TDengine](TDenginelogo.png)](https://www.taosdata.com)
 

cmake/define.inc

@@ -13,6 +13,18 @@ IF (TD_GRANT)
   ADD_DEFINITIONS(-D_GRANT)
 ENDIF ()
 
+IF (TD_SYNC)
+  ADD_DEFINITIONS(-D_SYNC)
+ENDIF ()
+
+IF (TD_MQTT)
+  ADD_DEFINITIONS(-D_MQTT)
+ENDIF ()
+
+IF (TD_TSDB_PLUGINS)
+  ADD_DEFINITIONS(-D_TSDB_PLUGINS)
+ENDIF ()
+
 IF (TD_GODLL)
   ADD_DEFINITIONS(-D_TD_GO_DLL_)
 ENDIF ()

cmake/input.inc

@@ -42,6 +42,16 @@ IF (${MEM_CHECK} MATCHES "true")
   MESSAGE(STATUS "build with memory check")
 ENDIF ()
 
+IF (${MQTT} MATCHES "false")
+  SET(TD_MQTT FALSE)
+  MESSAGE(STATUS "build without mqtt module")
+ENDIF ()
+
+IF (${SYNC} MATCHES "false")
+  SET(TD_SYNC FALSE)
+  MESSAGE(STATUS "build without sync module")
+ENDIF ()
+
 IF (${RANDOM_FILE_FAIL} MATCHES "true")
   SET(TD_RANDOM_FILE_FAIL TRUE)
   MESSAGE(STATUS "build with random-file-fail enabled")

cmake/platform.inc

@@ -114,6 +114,9 @@ ELSEIF (${OSTYPE} MATCHES "Ningsi80")
   MESSAGE(STATUS "input osType: Ningsi80")
 ELSEIF (${OSTYPE} MATCHES "Linux")
   MESSAGE(STATUS "input osType: Linux")
+ELSEIF (${OSTYPE} MATCHES "Alpine")
+  MESSAGE(STATUS "input osType: Alpine")
+  SET(TD_APLHINE TRUE)
 ELSE ()
   MESSAGE(STATUS "input osType unknown: " ${OSTYPE})
-ENDIF ()
\ No newline at end of file
+ENDIF ()

deps/CMakeLists.txt

@@ -10,6 +10,6 @@ ADD_SUBDIRECTORY(cJson)
 ADD_SUBDIRECTORY(wepoll)
 ADD_SUBDIRECTORY(MsvcLibX)
 
-IF (TD_LINUX)
+IF (TD_LINUX AND TD_MQTT)
   ADD_SUBDIRECTORY(MQTT-C)
 ENDIF ()
\ No newline at end of file

deps/MQTT-C/CMakeLists.txt

-cmake_minimum_required(VERSION 3.5)
-project(MQTT-C VERSION 1.1.2 LANGUAGES C)
+CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
 
 # MQTT-C build options
 option(MQTT_C_OpenSSL_SUPPORT "Build MQTT-C with OpenSSL support?" OFF)

documentation20/webdocs/markdowndocs/TAOS SQL-ch.md

@@ -82,13 +82,23 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
     ```
     删除数据库。所包含的全部数据表将被删除，谨慎使用
 
+- **修改数据库参数**
+    ```mysql
+    ALTER DATABASE db_name COMP 2;
+    ```
+    修改数据库文件压缩标志位，有效数字为0，1，2. 0表示不压缩，1表示一阶段压缩，2表示两阶段压缩。修改后可以使用show databases命令查看是否修改成功
+
+    ```mysql
+    ALTER DATABASE db_name REPLICA 2;
+    ```
+    修改数据库副本数，有效副本数为1到3。在集群中使用，副本数必须小于dnode的数目。修改后可以使用show databases命令查看是否修改成功
+
 
 - **显示系统所有数据库**
     ```mysql
     SHOW DATABASES;
     ```
 
-
 ## 表管理
 - **创建数据表**
   

documentation20/webdocs/markdowndocs/administrator-ch.md

@@ -47,6 +47,8 @@ Raw DataSize = numOfTables * rowSizePerTable * rowsPerTable
 
 因为TDengine具有很好的水平扩展能力，根据总量，再根据单个物理机或虚拟机的资源，就可以轻松决定需要购置多少台物理机或虚拟机了。
 
+**立即计算CPU、内存、存储，请参见：<a href='https://www.taosdata.com/config/config.html'>资源估算方法</a>**
+
 ## 容错和灾备
 
 ### 容错

documentation20/webdocs/markdowndocs/connector-java-ch.md

@@ -228,7 +228,8 @@ resultSet.close();
 stmt.close();
 conn.close();
 ```
-> `注意务必要将 connection 进行关闭`，否则会出现连接泄露。
+> `注意务必要将 connection 进行关闭`，否则会出现连接泄露。  
+
 ## 与连接池使用
 
 **HikariCP**

documentation20/webdocs/markdowndocs/faq-ch.md

@@ -19,7 +19,7 @@
 ## 4. 如何让TDengine crash时生成core文件？
 请看为此问题撰写的<a href='blog/2019/12/06/tdengine-crash时生成core文件的方法/'>技术博客</a>
 
-## 5. 遇到错误"Unable to establish connection", 我怎么办？
+## 5. 遇到错误"Unable to establish connection"或者"TDengine Error: Unable to resolve FQDN", 我怎么办？
 
 客户端遇到链接故障，请按照下面的步骤进行检查：
 

packaging/cfg/taos.cfg

@@ -122,11 +122,14 @@
 # number of replications, for cluster only 
 # replica               1
 
-# mqtt uri  
-# mqttBrokerAddress     mqtt://username:password@hostname:1883/taos/
+# mqtt hostname  
+# mqttHostName          test.mosquitto.org
 
-# mqtt client name 
-# mqttBrokerClientId    taos_mqtt
+# mqtt port  
+# mqttPort              1883
+
+# mqtt topic
+# mqttTopic             /test
 
 # the compressed rpc message, option:
 #  -1 (no compression)

snap/snapcraft.yaml

 name: tdengine
-base: core18 # the base snap is the execution environment for this snap
-version: '2.0.2.0' # just for humans, typically '1.2+git' or '1.3.2'
+base: core18
+version: 'RELEASE_VERSION'
 icon: snap/gui/t-dengine.svg
 summary: an open-source big data platform designed and optimized for IoT. 
 description: |
@@ -18,41 +18,18 @@ apps:
       - network
       - network-bind
       - system-observe
-      - systemfiles
 
   taos:
     command: taoswrapper.sh
     plugs:
       - network
       - system-observe
-      - systemfiles
-      - historyfile
 
   taosdemo:
     command: usr/bin/taosdemo
     plugs:
       - network
 
-plugs:
-  historyfile:
-    interface: personal-files
-    read:
-      - $HOME/.taos_history
-    write:
-      - $HOME/.taos_history
-
-  systemfiles:
-    interface: system-files
-    read:
-      - /etc/taos
-      - /var/lib/taos
-      - /var/log/taos
-      - /tmp
-    write:
-      - /var/log/taos
-      - /var/lib/taos
-      - /tmp
-
 parts:
   script:
     plugin: dump
@@ -95,7 +72,7 @@ parts:
       - usr/bin/taosd
       - usr/bin/taos
       - usr/bin/taosdemo
-      - usr/lib/libtaos.so.2.0.2.0
+      - usr/lib/libtaos.so.RELEASE_VERSION
       - usr/lib/libtaos.so.1
       - usr/lib/libtaos.so
 
@@ -115,8 +92,3 @@ layout:
     bind: $SNAP_DATA/var/log/taos
   /etc/taos:
     bind: $SNAP_DATA/etc/taos
-
-
-hooks:
-  install:
-    plugs: [systemfiles, historyfile]

src/CMakeLists.txt

@@ -10,7 +10,9 @@ ADD_SUBDIRECTORY(client)
 ADD_SUBDIRECTORY(query)
 ADD_SUBDIRECTORY(kit)
 ADD_SUBDIRECTORY(plugins)
-ADD_SUBDIRECTORY(sync)
+IF (TD_SYNC)
+  ADD_SUBDIRECTORY(sync)
+ENDIF ()
 ADD_SUBDIRECTORY(balance)
 ADD_SUBDIRECTORY(mnode)
 ADD_SUBDIRECTORY(vnode)

src/client/src/tscSQLParser.c

@@ -5088,7 +5088,7 @@ static int32_t setTimePrecision(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBIn
 }
 
 static void setCreateDBOption(SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
-  pMsg->maxTables = htonl(pCreateDb->maxTablesPerVnode);
+  pMsg->maxTables = htonl(-1);  // max tables can not be set anymore
   pMsg->cacheBlockSize = htonl(pCreateDb->cacheBlockSize);
   pMsg->totalBlocks = htonl(pCreateDb->numOfBlocks);
   pMsg->daysPerFile = htonl(pCreateDb->daysPerFile);

src/client/src/tscServer.c

@@ -406,7 +406,7 @@ int doProcessSql(SSqlObj *pSql) {
   if (code != TSDB_CODE_SUCCESS) {
     pRes->code = code;
     tscQueueAsyncRes(pSql);
-    return pRes->code;
+    return code;
   }
   
   return TSDB_CODE_SUCCESS;

src/client/src/tscUtil.c

@@ -718,6 +718,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) {
   return TSDB_CODE_SUCCESS;
 }
 
+// TODO: all subqueries should be freed correctly before close this connection.
 void tscCloseTscObj(STscObj* pObj) {
   assert(pObj != NULL);
   
@@ -727,6 +728,7 @@ void tscCloseTscObj(STscObj* pObj) {
   
   if (pObj->pDnodeConn != NULL) {
     rpcClose(pObj->pDnodeConn);
+    pObj->pDnodeConn = NULL;
   }
   
   tscDebug("%p DB connection is closed, dnodeConn:%p", pObj, pObj->pDnodeConn);

src/common/inc/tdataformat.h

@@ -325,8 +325,6 @@ void   tdResetKVRowBuilder(SKVRowBuilder *pBuilder);
 SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder);
 
 static FORCE_INLINE int tdAddColToKVRow(SKVRowBuilder *pBuilder, int16_t colId, int8_t type, void *value) {
-  ASSERT(pBuilder->nCols == 0 || colId > pBuilder->pColIdx[pBuilder->nCols - 1].colId);
-
   if (pBuilder->nCols >= pBuilder->tCols) {
     pBuilder->tCols *= 2;
     pBuilder->pColIdx = (SColIdx *)realloc((void *)(pBuilder->pColIdx), sizeof(SColIdx) * pBuilder->tCols);

src/common/inc/tglobal.h

@@ -104,8 +104,12 @@ extern int32_t  tsTelegrafUseFieldNum;
 
 // mqtt
 extern int32_t tsEnableMqttModule;
-extern char    tsMqttBrokerAddress[];
-extern char    tsMqttBrokerClientId[];
+extern char tsMqttHostName[];
+extern char tsMqttPort[];
+extern char tsMqttUser[];
+extern char tsMqttPass[];
+extern char tsMqttClientId[];
+extern char tsMqttTopic[];
 
 // monitor
 extern int32_t tsEnableMonitorModule;

src/common/src/tglobal.c

@@ -137,8 +137,12 @@ int32_t  tsTelegrafUseFieldNum = 0;
 
 // mqtt
 int32_t tsEnableMqttModule = 0;  // not finished yet, not started it by default
-char    tsMqttBrokerAddress[128] = {0};
-char    tsMqttBrokerClientId[128] = {0};
+char    tsMqttHostName[TSDB_MQTT_HOSTNAME_LEN] = "test.mosquitto.org";
+char    tsMqttPort[TSDB_MQTT_PORT_LEN] = "1883";
+char    tsMqttUser[TSDB_MQTT_USER_LEN] = {0};
+char    tsMqttPass[TSDB_MQTT_PASS_LEN] = {0};
+char    tsMqttClientId[TSDB_MQTT_CLIENT_ID_LEN] = "TDengineMqttSubscriber";
+char    tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test"; // #
 
 // monitor
 int32_t tsEnableMonitorModule = 1;
@@ -247,8 +251,11 @@ bool taosCfgDynamicOptions(char *msg) {
 
   for (int32_t i = 0; i < tsGlobalConfigNum; ++i) {
     SGlobalCfg *cfg = tsGlobalConfig + i;
-    if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
+    //if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
     if (cfg->valType != TAOS_CFG_VTYPE_INT32) continue;
+    
+    int32_t cfgLen = strlen(cfg->option);
+    if (cfgLen != olen) continue;
     if (strncasecmp(option, cfg->option, olen) != 0) continue;
     *((int32_t *)cfg->ptr) = vint;
 
@@ -767,26 +774,36 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "mqttBrokerAddress";
-  cfg.ptr = tsMqttBrokerAddress;
+  cfg.option = "mqttHostName";
+  cfg.ptr = tsMqttHostName;
   cfg.valType = TAOS_CFG_VTYPE_STRING;
-  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
   cfg.minValue = 0;
   cfg.maxValue = 0;
-  cfg.ptrLength = 126;
+  cfg.ptrLength = TSDB_MQTT_HOSTNAME_LEN;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "mqttBrokerClientId";
-  cfg.ptr = tsMqttBrokerClientId;
+  cfg.option = "mqttPort";
+  cfg.ptr = tsMqttPort;
   cfg.valType = TAOS_CFG_VTYPE_STRING;
-  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
   cfg.minValue = 0;
   cfg.maxValue = 0;
-  cfg.ptrLength = 126;
+  cfg.ptrLength = TSDB_MQTT_PORT_LEN;
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
- 
+
+  cfg.option = "mqttTopic";
+  cfg.ptr = tsMqttTopic;
+  cfg.valType = TAOS_CFG_VTYPE_STRING;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
+  cfg.minValue = 0;
+  cfg.maxValue = 0;
+  cfg.ptrLength = TSDB_MQTT_TOPIC_LEN;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "compressMsgSize";
   cfg.ptr = &tsCompressMsgSize;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
@@ -1270,6 +1287,9 @@ void taosInitGlobalCfg() {
 }
 
 bool taosCheckGlobalCfg() {
+  char fqdn[TSDB_FQDN_LEN];
+  uint16_t port;
+
   if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG || debugFlag & DEBUG_DUMP) {
     taosSetAllDebugFlag();
   }
@@ -1278,17 +1298,23 @@ bool taosCheckGlobalCfg() {
     taosGetFqdn(tsLocalFqdn);
   }
 
-  snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%d", tsLocalFqdn, tsServerPort);
+  snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
   uInfo("localEp is: %s", tsLocalEp);
 
   if (tsFirst[0] == 0) {
     strcpy(tsFirst, tsLocalEp);
+  } else {
+    taosGetFqdnPortFromEp(tsFirst, fqdn, &port);
+    snprintf(tsFirst, sizeof(tsFirst), "%s:%u", fqdn, port);
   }
 
   if (tsSecond[0] == 0) {
     strcpy(tsSecond, tsLocalEp);
+  } else {
+    taosGetFqdnPortFromEp(tsSecond, fqdn, &port);
+    snprintf(tsSecond, sizeof(tsSecond), "%s:%u", fqdn, port);
   }
-  
+
   taosGetSystemInfo();
 
   tsSetLocale();

src/dnode/CMakeLists.txt

@@ -11,10 +11,12 @@ AUX_SOURCE_DIRECTORY(src SRC)
 
 IF (TD_LINUX)
   ADD_EXECUTABLE(taosd ${SRC})
+  TARGET_LINK_LIBRARIES(taosd mnode monitor http tsdb twal vnode cJson lz4)
+
   IF (TD_SOMODE_STATIC)
-    TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http mqtt tsdb twal vnode cJson lz4 balance sync)
+    TARGET_LINK_LIBRARIES(taosd taos_static)
   ELSE ()
-    TARGET_LINK_LIBRARIES(taosd mnode taos monitor http mqtt tsdb twal vnode cJson lz4 balance sync)
+    TARGET_LINK_LIBRARIES(taosd taos)
   ENDIF ()
 
   IF (TD_ACCOUNT)
@@ -25,6 +27,14 @@ IF (TD_LINUX)
     TARGET_LINK_LIBRARIES(taosd grant)
   ENDIF ()
 
+  IF (TD_MQTT)
+    TARGET_LINK_LIBRARIES(taosd mqtt)
+  ENDIF ()
+
+  IF (TD_SYNC)
+    TARGET_LINK_LIBRARIES(taosd balance sync)
+  ENDIF ()
+
   SET(PREPARE_ENV_CMD "prepare_env_cmd")
   SET(PREPARE_ENV_TARGET "prepare_env_target")
   ADD_CUSTOM_COMMAND(OUTPUT ${PREPARE_ENV_CMD}

src/dnode/src/dnodeMgmt.c

@@ -611,7 +611,7 @@ static bool dnodeReadMnodeInfos() {
   }
 
   for (int i = 0; i < size; ++i) {
-    cJSON* nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
+    cJSON *nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
     if (nodeInfo == NULL) continue;
 
     cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId");
@@ -627,7 +627,7 @@ static bool dnodeReadMnodeInfos() {
       goto PARSE_OVER;
     }
     strncpy(tsDMnodeInfos.nodeInfos[i].nodeEp, nodeEp->valuestring, TSDB_EP_LEN);
- }
+  }
 
   ret = true;
 

src/dnode/src/dnodeModule.c

@@ -62,6 +62,7 @@ static void dnodeAllocModules() {
     dnodeSetModuleStatus(TSDB_MOD_HTTP);
   }
 
+#ifdef _MQTT
   tsModule[TSDB_MOD_MQTT].enable = (tsEnableMqttModule == 1);
   tsModule[TSDB_MOD_MQTT].name = "mqtt";
   tsModule[TSDB_MOD_MQTT].initFp = mqttInitSystem;
@@ -71,6 +72,7 @@ static void dnodeAllocModules() {
   if (tsEnableMqttModule) {
     dnodeSetModuleStatus(TSDB_MOD_MQTT);
   }
+#endif  
 
   tsModule[TSDB_MOD_MONITOR].enable    = (tsEnableMonitorModule == 1);
   tsModule[TSDB_MOD_MONITOR].name      = "monitor";

src/dnode/src/dnodeShell.c

@@ -154,15 +154,15 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
   rpcMsg.contLen = sizeof(SDMAuthMsg);
   rpcMsg.msgType = TSDB_MSG_TYPE_DM_AUTH;
   
-  dDebug("user:%s, send auth msg to mnode", user);
+  dDebug("user:%s, send auth msg to mnodes", user);
   SRpcMsg rpcRsp = {0};
   dnodeSendMsgToDnodeRecv(&rpcMsg, &rpcRsp);
 
   if (rpcRsp.code != 0) {
-    dError("user:%s, auth msg received from mnode, error:%s", user, tstrerror(rpcRsp.code));
+    dError("user:%s, auth msg received from mnodes, error:%s", user, tstrerror(rpcRsp.code));
   } else {
     SDMAuthRsp *pRsp = rpcRsp.pCont;
-    dDebug("user:%s, auth msg received from mnode", user);
+    dDebug("user:%s, auth msg received from mnodes", user);
     memcpy(secret, pRsp->secret, TSDB_KEY_LEN);
     memcpy(ckey, pRsp->ckey, TSDB_KEY_LEN);
     *spi = pRsp->spi;

src/inc/taosdef.h

@@ -272,6 +272,13 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_SHOW_SQL_LEN         64
 #define TSDB_SLOW_QUERY_SQL_LEN   512
 
+#define TSDB_MQTT_HOSTNAME_LEN    64
+#define TSDB_MQTT_PORT_LEN        8
+#define TSDB_MQTT_USER_LEN        24
+#define TSDB_MQTT_PASS_LEN        24
+#define TSDB_MQTT_TOPIC_LEN       64
+#define TSDB_MQTT_CLIENT_ID_LEN   32
+
 #define TSDB_METER_STATE_OFFLINE  0
 #define TSDB_METER_STATE_ONLLINE  1
 

src/inc/tmqtt.h

@@ -19,11 +19,11 @@
 #ifdef __cplusplus
 extern "C" {
 #endif
-#include <stdint.h>
+
 int32_t mqttInitSystem();
 int32_t mqttStartSystem();
-void mqttStopSystem();
-void mqttCleanUpSystem();
+void    mqttStopSystem();
+void    mqttCleanUpSystem();
 
 #ifdef __cplusplus
 }

src/inc/tsdb.h

@@ -38,6 +38,10 @@ extern "C" {
 #define TSDB_STATUS_COMMIT_START 1
 #define TSDB_STATUS_COMMIT_OVER 2
 
+// TSDB STATE DEFINITION
+#define TSDB_STATE_OK 0x0
+#define TSDB_STATE_BAD_FILE 0x1
+
 // --------- TSDB APPLICATION HANDLE DEFINITION
 typedef struct {
   void *appH;
@@ -80,6 +84,7 @@ int32_t      tsdbDropRepo(char *rootDir);
 TSDB_REPO_T *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH);
 void         tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit);
 int32_t      tsdbConfigRepo(TSDB_REPO_T *repo, STsdbCfg *pCfg);
+int          tsdbGetState(TSDB_REPO_T *repo);
 
 // --------- TSDB TABLE DEFINITION
 typedef struct {

src/inc/ttokendef.h

@@ -101,127 +101,126 @@
 #define TK_CONNS                           83
 #define TK_STATE                           84
 #define TK_KEEP                            85
-#define TK_MAXTABLES                       86
-#define TK_CACHE                           87
-#define TK_REPLICA                         88
-#define TK_QUORUM                          89
-#define TK_DAYS                            90
-#define TK_MINROWS                         91
-#define TK_MAXROWS                         92
-#define TK_BLOCKS                          93
-#define TK_CTIME                           94
-#define TK_WAL                             95
-#define TK_FSYNC                           96
-#define TK_COMP                            97
-#define TK_PRECISION                       98
-#define TK_LP                              99
-#define TK_RP                             100
-#define TK_TAGS                           101
-#define TK_USING                          102
-#define TK_AS                             103
-#define TK_COMMA                          104
-#define TK_NULL                           105
-#define TK_SELECT                         106
-#define TK_UNION                          107
-#define TK_ALL                            108
-#define TK_FROM                           109
-#define TK_VARIABLE                       110
-#define TK_INTERVAL                       111
-#define TK_FILL                           112
-#define TK_SLIDING                        113
-#define TK_ORDER                          114
-#define TK_BY                             115
-#define TK_ASC                            116
-#define TK_DESC                           117
-#define TK_GROUP                          118
-#define TK_HAVING                         119
-#define TK_LIMIT                          120
-#define TK_OFFSET                         121
-#define TK_SLIMIT                         122
-#define TK_SOFFSET                        123
-#define TK_WHERE                          124
-#define TK_NOW                            125
-#define TK_RESET                          126
-#define TK_QUERY                          127
-#define TK_ADD                            128
-#define TK_COLUMN                         129
-#define TK_TAG                            130
-#define TK_CHANGE                         131
-#define TK_SET                            132
-#define TK_KILL                           133
-#define TK_CONNECTION                     134
-#define TK_STREAM                         135
-#define TK_COLON                          136
-#define TK_ABORT                          137
-#define TK_AFTER                          138
-#define TK_ATTACH                         139
-#define TK_BEFORE                         140
-#define TK_BEGIN                          141
-#define TK_CASCADE                        142
-#define TK_CLUSTER                        143
-#define TK_CONFLICT                       144
-#define TK_COPY                           145
-#define TK_DEFERRED                       146
-#define TK_DELIMITERS                     147
-#define TK_DETACH                         148
-#define TK_EACH                           149
-#define TK_END                            150
-#define TK_EXPLAIN                        151
-#define TK_FAIL                           152
-#define TK_FOR                            153
-#define TK_IGNORE                         154
-#define TK_IMMEDIATE                      155
-#define TK_INITIALLY                      156
-#define TK_INSTEAD                        157
-#define TK_MATCH                          158
-#define TK_KEY                            159
-#define TK_OF                             160
-#define TK_RAISE                          161
-#define TK_REPLACE                        162
-#define TK_RESTRICT                       163
-#define TK_ROW                            164
-#define TK_STATEMENT                      165
-#define TK_TRIGGER                        166
-#define TK_VIEW                           167
-#define TK_COUNT                          168
-#define TK_SUM                            169
-#define TK_AVG                            170
-#define TK_MIN                            171
-#define TK_MAX                            172
-#define TK_FIRST                          173
-#define TK_LAST                           174
-#define TK_TOP                            175
-#define TK_BOTTOM                         176
-#define TK_STDDEV                         177
-#define TK_PERCENTILE                     178
-#define TK_APERCENTILE                    179
-#define TK_LEASTSQUARES                   180
-#define TK_HISTOGRAM                      181
-#define TK_DIFF                           182
-#define TK_SPREAD                         183
-#define TK_TWA                            184
-#define TK_INTERP                         185
-#define TK_LAST_ROW                       186
-#define TK_RATE                           187
-#define TK_IRATE                          188
-#define TK_SUM_RATE                       189
-#define TK_SUM_IRATE                      190
-#define TK_AVG_RATE                       191
-#define TK_AVG_IRATE                      192
-#define TK_TBID                           193
-#define TK_SEMI                           194
-#define TK_NONE                           195
-#define TK_PREV                           196
-#define TK_LINEAR                         197
-#define TK_IMPORT                         198
-#define TK_METRIC                         199
-#define TK_TBNAME                         200
-#define TK_JOIN                           201
-#define TK_METRICS                        202
-#define TK_STABLE                         203
-#define TK_INSERT                         204
-#define TK_INTO                           205
-#define TK_VALUES                         206
+#define TK_CACHE                           86
+#define TK_REPLICA                         87
+#define TK_QUORUM                          88
+#define TK_DAYS                            89
+#define TK_MINROWS                         90
+#define TK_MAXROWS                         91
+#define TK_BLOCKS                          92
+#define TK_CTIME                           93
+#define TK_WAL                             94
+#define TK_FSYNC                           95
+#define TK_COMP                            96
+#define TK_PRECISION                       97
+#define TK_LP                              98
+#define TK_RP                              99
+#define TK_TAGS                           100
+#define TK_USING                          101
+#define TK_AS                             102
+#define TK_COMMA                          103
+#define TK_NULL                           104
+#define TK_SELECT                         105
+#define TK_UNION                          106
+#define TK_ALL                            107
+#define TK_FROM                           108
+#define TK_VARIABLE                       109
+#define TK_INTERVAL                       110
+#define TK_FILL                           111
+#define TK_SLIDING                        112
+#define TK_ORDER                          113
+#define TK_BY                             114
+#define TK_ASC                            115
+#define TK_DESC                           116
+#define TK_GROUP                          117
+#define TK_HAVING                         118
+#define TK_LIMIT                          119
+#define TK_OFFSET                         120
+#define TK_SLIMIT                         121
+#define TK_SOFFSET                        122
+#define TK_WHERE                          123
+#define TK_NOW                            124
+#define TK_RESET                          125
+#define TK_QUERY                          126
+#define TK_ADD                            127
+#define TK_COLUMN                         128
+#define TK_TAG                            129
+#define TK_CHANGE                         130
+#define TK_SET                            131
+#define TK_KILL                           132
+#define TK_CONNECTION                     133
+#define TK_STREAM                         134
+#define TK_COLON                          135
+#define TK_ABORT                          136
+#define TK_AFTER                          137
+#define TK_ATTACH                         138
+#define TK_BEFORE                         139
+#define TK_BEGIN                          140
+#define TK_CASCADE                        141
+#define TK_CLUSTER                        142
+#define TK_CONFLICT                       143
+#define TK_COPY                           144
+#define TK_DEFERRED                       145
+#define TK_DELIMITERS                     146
+#define TK_DETACH                         147
+#define TK_EACH                           148
+#define TK_END                            149
+#define TK_EXPLAIN                        150
+#define TK_FAIL                           151
+#define TK_FOR                            152
+#define TK_IGNORE                         153
+#define TK_IMMEDIATE                      154
+#define TK_INITIALLY                      155
+#define TK_INSTEAD                        156
+#define TK_MATCH                          157
+#define TK_KEY                            158
+#define TK_OF                             159
+#define TK_RAISE                          160
+#define TK_REPLACE                        161
+#define TK_RESTRICT                       162
+#define TK_ROW                            163
+#define TK_STATEMENT                      164
+#define TK_TRIGGER                        165
+#define TK_VIEW                           166
+#define TK_COUNT                          167
+#define TK_SUM                            168
+#define TK_AVG                            169
+#define TK_MIN                            170
+#define TK_MAX                            171
+#define TK_FIRST                          172
+#define TK_LAST                           173
+#define TK_TOP                            174
+#define TK_BOTTOM                         175
+#define TK_STDDEV                         176
+#define TK_PERCENTILE                     177
+#define TK_APERCENTILE                    178
+#define TK_LEASTSQUARES                   179
+#define TK_HISTOGRAM                      180
+#define TK_DIFF                           181
+#define TK_SPREAD                         182
+#define TK_TWA                            183
+#define TK_INTERP                         184
+#define TK_LAST_ROW                       185
+#define TK_RATE                           186
+#define TK_IRATE                          187
+#define TK_SUM_RATE                       188
+#define TK_SUM_IRATE                      189
+#define TK_AVG_RATE                       190
+#define TK_AVG_IRATE                      191
+#define TK_TBID                           192
+#define TK_SEMI                           193
+#define TK_NONE                           194
+#define TK_PREV                           195
+#define TK_LINEAR                         196
+#define TK_IMPORT                         197
+#define TK_METRIC                         198
+#define TK_TBNAME                         199
+#define TK_JOIN                           200
+#define TK_METRICS                        201
+#define TK_STABLE                         202
+#define TK_INSERT                         203
+#define TK_INTO                           204
+#define TK_VALUES                         205
 
 
 #define TK_SPACE                          300

src/kit/shell/src/shellLinux.c

@@ -409,7 +409,7 @@ void set_terminal_mode() {
   }
 }
 
-void get_history_path(char *history) { sprintf(history, "%s/%s", getpwuid(getuid())->pw_dir, HISTORY_FILE); }
+void get_history_path(char *history) { sprintf(history, "%s/%s", getenv("HOME"), HISTORY_FILE); }
 
 void clearScreen(int ecmd_pos, int cursor_pos) {
   struct winsize w;

src/mnode/src/mnodeDb.c

@@ -242,6 +242,7 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
+#if 0
   if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > pCfg->daysToKeep) {
     mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep, TSDB_MIN_KEEP, pCfg->daysToKeep);
     return TSDB_CODE_MND_INVALID_DB_OPTION;
@@ -251,6 +252,7 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
     mError("invalid db option daysToKeep1:%d valid range: [%d, %d]", pCfg->daysToKeep1, TSDB_MIN_KEEP, pCfg->daysToKeep2);
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
+#endif
 
   if (pCfg->maxRowsPerFileBlock < TSDB_MIN_MAX_ROW_FBLOCK || pCfg->maxRowsPerFileBlock > TSDB_MAX_MAX_ROW_FBLOCK) {
     mError("invalid db option maxRowsPerFileBlock:%d valid range: [%d, %d]", pCfg->maxRowsPerFileBlock,
@@ -310,6 +312,13 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
     return TSDB_CODE_MND_INVALID_DB_OPTION;
   }
 
+#ifndef _SYNC
+  if (pCfg->replications != 1) {
+    mError("invalid db option replications:%d can only be 1 in this version", pCfg->replications);
+    return TSDB_CODE_MND_INVALID_DB_OPTION;
+  }
+#endif
+
   return TSDB_CODE_SUCCESS;
 }
 

src/mnode/src/mnodeDnode.c

@@ -78,6 +78,9 @@ static int32_t mnodeDnodeActionInsert(SSdbOper *pOper) {
 static int32_t mnodeDnodeActionDelete(SSdbOper *pOper) {
   SDnodeObj *pDnode = pOper->pObj;
  
+#ifndef _SYNC 
+  mnodeDropAllDnodeVgroups(pDnode);
+#endif  
   mnodeDropMnodeLocal(pDnode->dnodeId);
   balanceAsyncNotify();
 
@@ -585,7 +588,11 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
 
   mInfo("dnode:%d, start to drop it", pDnode->dnodeId);
 
+#ifndef _SYNC
+  int32_t code = mnodeDropDnode(pDnode, pMsg);
+#else
   int32_t code = balanceDropDnode(pDnode);
+#endif  
   mnodeDecDnodeRef(pDnode);
   return code;
 }
@@ -1043,3 +1050,59 @@ static char* mnodeGetDnodeAlternativeRoleStr(int32_t alternativeRole) {
   }
 }
 
+#ifndef _SYNC
+
+int32_t balanceInit() { return TSDB_CODE_SUCCESS; }
+void    balanceCleanUp() {}
+void    balanceAsyncNotify() {}
+void    balanceSyncNotify() {}
+void    balanceReset() {}
+int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; }
+
+char* syncRole[] = {
+  "offline",
+  "unsynced",
+  "syncing",
+  "slave",
+  "master"
+};
+
+int32_t balanceAllocVnodes(SVgObj *pVgroup) {
+  void *     pIter = NULL;
+  SDnodeObj *pDnode = NULL;
+  SDnodeObj *pSelDnode = NULL;
+  float      vnodeUsage = 1000.0;
+
+  while (1) {
+    pIter = mnodeGetNextDnode(pIter, &pDnode);
+    if (pDnode == NULL) break;
+
+    if (pDnode->numOfCores > 0 && pDnode->openVnodes < TSDB_MAX_VNODES) {
+      float openVnodes = pDnode->openVnodes;
+      if (pDnode->isMgmt) openVnodes += tsMnodeEqualVnodeNum;
+
+      float usage = openVnodes / pDnode->numOfCores;
+      if (usage <= vnodeUsage) {
+        pSelDnode = pDnode;
+        vnodeUsage = usage;
+      }
+    }
+    mnodeDecDnodeRef(pDnode);
+  }
+
+  sdbFreeIter(pIter);
+
+  if (pSelDnode == NULL) {
+    mError("failed to alloc vnode to vgroup");
+    return TSDB_CODE_MND_NO_ENOUGH_DNODES;
+  }
+
+  pVgroup->vnodeGid[0].dnodeId = pSelDnode->dnodeId;
+  pVgroup->vnodeGid[0].pDnode = pSelDnode;
+
+  mDebug("dnode:%d, alloc one vnode to vgroup, openVnodes:%d", pSelDnode->dnodeId, pSelDnode->openVnodes);
+  return TSDB_CODE_SUCCESS;
+}
+
+#endif 
+

src/mnode/src/mnodeProfile.c

@@ -98,8 +98,10 @@ SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
     .connId = connId,
     .stime  = taosGetTimestampMs()
   };
+
   tstrncpy(connObj.user, user, sizeof(connObj.user));
-  
+  connObj.lastAccess = connObj.stime;
+
   SConnObj *pConn = taosCachePut(tsMnodeConnCache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), CONN_KEEP_TIME * 1000);
 
   mDebug("connId:%d, is created, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
@@ -244,6 +246,7 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    if (pConnObj->lastAccess < pConnObj->stime) pConnObj->lastAccess = pConnObj->stime;
     *(int64_t *)pWrite = pConnObj->lastAccess;
     cols++;
 

src/mnode/src/mnodeTable.c

@@ -1711,14 +1711,20 @@ static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
       mnodeDestroyChildTable(pTable);
       return TSDB_CODE_MND_INVALID_TABLE_NAME;
     }
-    
+
     pTable->suid = pMsg->pSTable->uid;
-    pTable->uid  = (((uint64_t)pTable->vgId) << 40) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
-                  (sdbGetVersion() & ((1ul << 16) - 1ul));
+    pTable->uid = (((uint64_t)pTable->vgId) << 48) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 24) +
+                  ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
     pTable->superTable = pMsg->pSTable;
   } else {
-    pTable->uid = (((uint64_t)pTable->vgId) << 40) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
-                  (sdbGetVersion() & ((1ul << 16) - 1ul));
+    if (pTable->info.type == TSDB_SUPER_TABLE) {
+      int64_t us = taosGetTimestampUs();
+      pTable->uid = (us << 24) + ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
+    } else {
+      pTable->uid = (((uint64_t)pTable->vgId) << 48) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 24) +
+                    ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
+    }
+
     pTable->sversion     = 0;
     pTable->numOfColumns = htons(pCreate->numOfColumns);
     pTable->sqlLen       = htons(pCreate->sqlLen);

src/mnode/src/mnodeUser.c

@@ -581,7 +581,7 @@ void mnodeDropAllUsers(SAcctObj *pAcct)  {
 int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
   if (!sdbIsMaster()) {
     *secret = 0;
-    mDebug("user:%s, failed to auth user, reason:%s", user, tstrerror(TSDB_CODE_APP_NOT_READY));
+    mDebug("user:%s, failed to auth user, mnode is not master", user);
     return TSDB_CODE_APP_NOT_READY;
   }
 

src/os/inc/os.h

@@ -36,7 +36,7 @@ extern "C" {
 #include "osLinux32.h"
 #endif
 
-#ifdef _TD_ALPINE
+#ifdef _ALPINE
 #include "osAlpine.h"
 #endif
 

src/os/inc/osAlpine.h

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_OS_LINUX64_H
-#define TDENGINE_OS_LINUX64_H
+#ifndef TDENGINE_OS_ALPINE_H
+#define TDENGINE_OS_ALPINE_H
 
 #ifdef __cplusplus
 extern "C" {

src/os/inc/osLinux64.h

@@ -75,7 +75,9 @@ extern "C" {
 #include <fcntl.h>
 #include <sys/utsname.h>
 #include <sys/resource.h>
+#ifndef _ALPINE
 #include <error.h>
+#endif
 #include <linux/sysctl.h>
 
 #ifdef __cplusplus

src/os/src/detail/osSysinfo.c

@@ -579,7 +579,11 @@ void taosSetCoreDump() {
   struct rlimit rlim;
   struct rlimit rlim_new;
   if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
+    #ifndef _ALPINE
     uInfo("the old unlimited para: rlim_cur=%" PRIu64 ", rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
+    #else
+    uInfo("the old unlimited para: rlim_cur=%llu, rlim_max=%llu", rlim.rlim_cur, rlim.rlim_max);
+    #endif
     rlim_new.rlim_cur = RLIM_INFINITY;
     rlim_new.rlim_max = RLIM_INFINITY;
     if (setrlimit(RLIMIT_CORE, &rlim_new) != 0) {
@@ -591,7 +595,11 @@ void taosSetCoreDump() {
   }
 
   if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
+    #ifndef _ALPINE
     uInfo("the new unlimited para: rlim_cur=%" PRIu64 ", rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
+    #else
+    uInfo("the new unlimited para: rlim_cur=%llu, rlim_max=%llu", rlim.rlim_cur, rlim.rlim_max);
+    #endif
   }
 
 #ifndef _TD_ARM_
@@ -659,4 +667,4 @@ bool taosGetSystemUid(char *uid) {
   return false;
 }
 
-#endif
\ No newline at end of file
+#endif

src/plugins/CMakeLists.txt

@@ -3,4 +3,6 @@ PROJECT(TDengine)
 
 ADD_SUBDIRECTORY(monitor)
 ADD_SUBDIRECTORY(http)
-ADD_SUBDIRECTORY(mqtt)
+IF (TD_MQTT)
+  ADD_SUBDIRECTORY(mqtt)
+ENDIF ()
\ No newline at end of file

src/plugins/http/CMakeLists.txt

@@ -11,11 +11,12 @@ AUX_SOURCE_DIRECTORY(src SRC)
   
 IF (TD_LINUX)
   ADD_LIBRARY(http ${SRC})
+  TARGET_LINK_LIBRARIES(http z)
 
   IF (TD_SOMODE_STATIC)
-    TARGET_LINK_LIBRARIES(http taos_static z)
+    TARGET_LINK_LIBRARIES(http taos_static)
   ELSE ()  
-    TARGET_LINK_LIBRARIES(http taos z)
+    TARGET_LINK_LIBRARIES(http taos)
   ENDIF ()
 
   IF (TD_ADMIN)

src/plugins/monitor/CMakeLists.txt

@@ -2,11 +2,11 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
 PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
 AUX_SOURCE_DIRECTORY(./src SRC)
-  
+
 IF (TD_LINUX)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
   ADD_LIBRARY(monitor ${SRC})
   
   IF (TD_SOMODE_STATIC)

src/plugins/monitor/src/monitorMain.c

@@ -25,6 +25,7 @@
 #include "tsclient.h"
 #include "dnode.h"
 #include "monitor.h"
+#include "taoserror.h"
 
 #define monitorFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
 #define monitorError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
@@ -33,129 +34,159 @@
 #define monitorDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
 #define monitorTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
 
-#define SQL_LENGTH     1024
+#define SQL_LENGTH     1030
 #define LOG_LEN_STR    100
 #define IP_LEN_STR     TSDB_EP_LEN
 #define CHECK_INTERVAL 1000
 
 typedef enum {
-  MONITOR_CMD_CREATE_DB,
-  MONITOR_CMD_CREATE_TB_LOG,
-  MONITOR_CMD_CREATE_MT_DN,
-  MONITOR_CMD_CREATE_MT_ACCT,
-  MONITOR_CMD_CREATE_TB_DN,
-  MONITOR_CMD_CREATE_TB_ACCT_ROOT,
-  MONITOR_CMD_CREATE_TB_SLOWQUERY,
-  MONITOR_CMD_MAX
+  MON_CMD_CREATE_DB,
+  MON_CMD_CREATE_TB_LOG,
+  MON_CMD_CREATE_MT_DN,
+  MON_CMD_CREATE_MT_ACCT,
+  MON_CMD_CREATE_TB_DN,
+  MON_CMD_CREATE_TB_ACCT_ROOT,
+  MON_CMD_CREATE_TB_SLOWQUERY,
+  MON_CMD_MAX
 } EMonitorCommand;
 
 typedef enum {
-  MONITOR_STATE_UN_INIT,
-  MONITOR_STATE_INITIALIZING,
-  MONITOR_STATE_INITIALIZED,
-  MONITOR_STATE_STOPPED
+  MON_STATE_NOT_INIT,
+  MON_STATE_INITED
 } EMonitorState;
 
 typedef struct {
-  void * conn;
-  void * timer;
-  char   ep[TSDB_EP_LEN];
-  int8_t cmdIndex;
-  int8_t state;
-  char   sql[SQL_LENGTH + 1];
-  void * initTimer;
-  void * diskTimer;
+  pthread_t thread;
+  void *    conn;
+  char      ep[TSDB_EP_LEN];
+  int8_t    cmdIndex;
+  int8_t    state;
+  int8_t    start;   // enable/disable by mnode
+  int8_t    quiting; // taosd is quiting 
+  char      sql[SQL_LENGTH + 1];
 } SMonitorConn;
 
-static SMonitorConn tsMonitorConn;
-static void monitorInitConn(void *para, void *unused);
-static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code);
-static void monitorInitDatabase();
-static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code);
-static void monitorStartTimer();
-static void monitorSaveSystemInfo();
+static SMonitorConn tsMonitor = {0};
+static void  monitorSaveSystemInfo();
+static void *monitorThreadFunc(void *param);
+static void  monitorBuildMonitorSql(char *sql, int32_t cmd);
 extern int32_t (*monitorStartSystemFp)();
-extern void (*monitorStopSystemFp)();
-extern void (*monitorExecuteSQLFp)(char *sql);
-
-static void monitorCheckDiskUsage(void *para, void *unused) {
-  taosGetDisk();
-  taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
-}
+extern void    (*monitorStopSystemFp)();
+extern void    (*monitorExecuteSQLFp)(char *sql);
 
 int32_t monitorInitSystem() {
-  taos_init();
-  taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
+  if (tsMonitor.ep[0] == 0) {
+    strcpy(tsMonitor.ep, tsLocalEp);
+  }
+
+  int len = strlen(tsMonitor.ep);
+  for (int i = 0; i < len; ++i) {
+    if (tsMonitor.ep[i] == ':' || tsMonitor.ep[i] == '-' || tsMonitor.ep[i] == '.') {
+      tsMonitor.ep[i] = '_';
+    }
+  }
+
+  pthread_attr_t thAttr;
+  pthread_attr_init(&thAttr);
+  pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
+
+  if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) {
+    monitorError("failed to create thread to for monitor module, reason:%s", strerror(errno));
+    return -1;
+  }
+
+  pthread_attr_destroy(&thAttr);
+  monitorDebug("monitor thread is launched");
+
   monitorStartSystemFp = monitorStartSystem;
   monitorStopSystemFp = monitorStopSystem;
   return 0;
 }
 
 int32_t monitorStartSystem() {
-  monitorInfo("start monitor module");
-  monitorInitSystem();
-  taosTmrReset(monitorInitConn, 10, NULL, tscTmr, &tsMonitorConn.initTimer);
+  taos_init();
+  tsMonitor.start = 1;
+  monitorExecuteSQLFp = monitorExecuteSQL;
+  monitorInfo("monitor module start");
   return 0;
 }
 
-static void monitorStartSystemRetry() {
-  if (tsMonitorConn.initTimer != NULL) {
-    taosTmrReset(monitorInitConn, 3000, NULL, tscTmr, &tsMonitorConn.initTimer);
-  }
-}
+static void *monitorThreadFunc(void *param) {
+  monitorDebug("starting to initialize monitor module ...");
 
-static void monitorInitConn(void *para, void *unused) {
-  if (dnodeGetDnodeId() <= 0) {
-    monitorStartSystemRetry();
-    return;
-  }
-  
-  monitorInfo("starting to initialize monitor service ..");
-  tsMonitorConn.state = MONITOR_STATE_INITIALIZING;
+  while (1) {
+    if (tsMonitor.quiting) {
+      tsMonitor.state = MON_STATE_NOT_INIT;
+      monitorInfo("monitor thread will quit, for taosd is quiting");
+      break;
+    } else {
+      taosGetDisk();
+    }
 
-  if (tsMonitorConn.ep[0] == 0) 
-    strcpy(tsMonitorConn.ep, tsLocalEp);
+    if (tsMonitor.start == 0) {
+      continue;
+    }
 
-  int len = strlen(tsMonitorConn.ep);
-  for (int i = 0; i < len; ++i) {
-    if (tsMonitorConn.ep[i] == ':' || tsMonitorConn.ep[i] == '-') {
-      tsMonitorConn.ep[i] = '_';
+    static int32_t accessTimes = 0;
+    accessTimes++;
+    taosMsleep(1000);
+
+    if (dnodeGetDnodeId() <= 0) {
+      monitorDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
+      continue;
     }
-  }
 
-  if (tsMonitorConn.conn == NULL) {
-    taos_connect_a(NULL, "monitor", tsInternalPass, "", 0, monitorInitConnCb, &tsMonitorConn, &(tsMonitorConn.conn));
-  } else {
-    monitorInitDatabase();
-  }
-}
+    if (tsMonitor.conn == NULL) {
+      tsMonitor.state = MON_STATE_NOT_INIT;
+      tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0);
+      if (tsMonitor.conn == NULL) {
+        monitorError("failed to connect to database, reason:%s", tstrerror(terrno));
+        continue;
+      } else {
+        monitorDebug("connect to database success");
+      }
+    }
 
-static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code) {
-  // free it firstly in any cases.
-  taos_free_result(result);
+    if (tsMonitor.state == MON_STATE_NOT_INIT) {
+      for (; tsMonitor.cmdIndex < MON_CMD_MAX; ++tsMonitor.cmdIndex) {
+        monitorBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex);
+        void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
+        int   code = taos_errno(res);
+        taos_free_result(res);
+
+        if (code != 0) {
+          monitorError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
+          break;
+        } else {
+          monitorDebug("successfully to exec sql:%s", tsMonitor.sql);
+        }
+      }
+
+      if (tsMonitor.start) {
+        tsMonitor.state = MON_STATE_INITED;
+      }
+    }
 
-  if (code != TSDB_CODE_SUCCESS) {
-    monitorError("monitor:%p, connect to database failed, reason:%s", tsMonitorConn.conn, tstrerror(code));
-    taos_close(tsMonitorConn.conn);
-    tsMonitorConn.conn = NULL;
-    tsMonitorConn.state = MONITOR_STATE_UN_INIT;
-    monitorStartSystemRetry();
-    return;
+    if (tsMonitor.state == MON_STATE_INITED) {
+      if (accessTimes % tsMonitorInterval == 0) {
+        monitorSaveSystemInfo();
+      }
+    }
   }
 
-  monitorDebug("monitor:%p, connect to database success, reason:%s", tsMonitorConn.conn, tstrerror(code));
-  monitorInitDatabase();
+  monitorInfo("monitor thread is stopped");
+  return NULL;
 }
 
-static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
+static void monitorBuildMonitorSql(char *sql, int32_t cmd) {
   memset(sql, 0, SQL_LENGTH);
 
-  if (cmd == MONITOR_CMD_CREATE_DB) {
+  if (cmd == MON_CMD_CREATE_DB) {
     snprintf(sql, SQL_LENGTH,
              "create database if not exists %s replica 1 days 10 keep 30 cache %d "
-             "blocks %d maxtables 16 precision 'us'",
+             "blocks %d precision 'us'",
              tsMonitorDbName, TSDB_MIN_CACHE_BLOCK_SIZE, TSDB_MIN_TOTAL_BLOCKS);
-  } else if (cmd == MONITOR_CMD_CREATE_MT_DN) {
+  } else if (cmd == MON_CMD_CREATE_MT_DN) {
     snprintf(sql, SQL_LENGTH,
              "create table if not exists %s.dn(ts timestamp"
              ", cpu_taosd float, cpu_system float, cpu_cores int"
@@ -166,10 +197,10 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
              ", req_http int, req_select int, req_insert int"
              ") tags (dnodeid int, fqdn binary(%d))",
              tsMonitorDbName, TSDB_FQDN_LEN);
-  } else if (cmd == MONITOR_CMD_CREATE_TB_DN) {
+  } else if (cmd == MON_CMD_CREATE_TB_DN) {
     snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn%d using %s.dn tags(%d, '%s')", tsMonitorDbName,
              dnodeGetDnodeId(), tsMonitorDbName, dnodeGetDnodeId(), tsLocalEp);
-  } else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) {
+  } else if (cmd == MON_CMD_CREATE_MT_ACCT) {
     snprintf(sql, SQL_LENGTH,
              "create table if not exists %s.acct(ts timestamp "
              ", currentPointsPerSecond bigint, maxPointsPerSecond bigint"
@@ -185,15 +216,15 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
              ", accessState smallint"
              ") tags (acctId binary(%d))",
              tsMonitorDbName, TSDB_USER_LEN);
-  } else if (cmd == MONITOR_CMD_CREATE_TB_ACCT_ROOT) {
+  } else if (cmd == MON_CMD_CREATE_TB_ACCT_ROOT) {
     snprintf(sql, SQL_LENGTH, "create table if not exists %s.acct_%s using %s.acct tags('%s')", tsMonitorDbName, TSDB_DEFAULT_USER,
              tsMonitorDbName, TSDB_DEFAULT_USER);
-  } else if (cmd == MONITOR_CMD_CREATE_TB_SLOWQUERY) {
+  } else if (cmd == MON_CMD_CREATE_TB_SLOWQUERY) {
     snprintf(sql, SQL_LENGTH,
              "create table if not exists %s.slowquery(ts timestamp, username "
              "binary(%d), created_time timestamp, time bigint, sql binary(%d))",
              tsMonitorDbName, TSDB_TABLE_FNAME_LEN - 1, TSDB_SLOW_QUERY_SQL_LEN);
-  } else if (cmd == MONITOR_CMD_CREATE_TB_LOG) {
+  } else if (cmd == MON_CMD_CREATE_TB_LOG) {
     snprintf(sql, SQL_LENGTH,
              "create table if not exists %s.log(ts timestamp, level tinyint, "
              "content binary(%d), ipaddr binary(%d))",
@@ -203,75 +234,22 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
   sql[SQL_LENGTH] = 0;
 }
 
-static void monitorInitDatabase() {
-  if (tsMonitorConn.cmdIndex < MONITOR_CMD_MAX) {
-    dnodeBuildMonitorSql(tsMonitorConn.sql, tsMonitorConn.cmdIndex);
-    taos_query_a(tsMonitorConn.conn, tsMonitorConn.sql, monitorInitDatabaseCb, NULL);
-  } else {
-    tsMonitorConn.state = MONITOR_STATE_INITIALIZED;
-    monitorExecuteSQLFp = monitorExecuteSQL;
-    monitorInfo("monitor service init success");
-
-    monitorStartTimer();
-  }
-}
-
-static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code) {
-  if (code == TSDB_CODE_MND_TABLE_ALREADY_EXIST || code == TSDB_CODE_MND_DB_ALREADY_EXIST || code >= 0) {
-    monitorDebug("monitor:%p, sql success, reason:%s, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
-    if (tsMonitorConn.cmdIndex == MONITOR_CMD_CREATE_TB_LOG) {
-      monitorInfo("dnode:%s is started", tsLocalEp);
-    }
-    tsMonitorConn.cmdIndex++;
-    monitorInitDatabase();
-  } else {
-    monitorError("monitor:%p, sql failed, reason:%s, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
-    tsMonitorConn.state = MONITOR_STATE_UN_INIT;
-    monitorStartSystemRetry();
-  }
-
-  taos_free_result(result);
-}
-
 void monitorStopSystem() {
-  if (tsMonitorConn.state == MONITOR_STATE_STOPPED) return;
-  tsMonitorConn.state = MONITOR_STATE_STOPPED;
+  tsMonitor.start = 0;
+  tsMonitor.state = MON_STATE_NOT_INIT;
   monitorExecuteSQLFp = NULL;
-
-  monitorInfo("monitor module is stopped");
-
-  if (tsMonitorConn.initTimer != NULL) {
-    taosTmrStopA(&(tsMonitorConn.initTimer));
-  }
-  if (tsMonitorConn.timer != NULL) {
-    taosTmrStopA(&(tsMonitorConn.timer));
-  }
-  if (tsMonitorConn.conn != NULL) {
-    taos_close(tsMonitorConn.conn);
-    tsMonitorConn.conn = NULL;
-  }
+  monitorInfo("monitor module stopped");
 }
 
 void monitorCleanUpSystem() {
+  tsMonitor.quiting = 1;
   monitorStopSystem();
-  monitorInfo("monitor module cleanup");
-}
-
-static void monitorStartTimer() {
-  taosTmrReset(monitorSaveSystemInfo, tsMonitorInterval * 1000, NULL, tscTmr, &tsMonitorConn.timer);
-}
-
-static void dnodeMontiorLogCallback(void *param, TAOS_RES *result, int32_t code) {
-  int32_t c = taos_errno(result);
-
-  if (c != TSDB_CODE_SUCCESS) {
-    monitorError("monitor:%p, save %s failed, reason:%s", tsMonitorConn.conn, (char *)param, tstrerror(c));
-  } else {
-    int32_t rows = taos_affected_rows(result);
-    monitorDebug("monitor:%p, save %s succ, rows:%d", tsMonitorConn.conn, (char *)param, rows);
+  pthread_join(tsMonitor.thread, NULL);
+  if (tsMonitor.conn != NULL) {
+    taos_close(tsMonitor.conn);
+    tsMonitor.conn = NULL;
   }
-
-  taos_free_result(result);
+  monitorInfo("monitor module is cleaned up");
 }
 
 // unit is MB
@@ -279,13 +257,13 @@ static int32_t monitorBuildMemorySql(char *sql) {
   float sysMemoryUsedMB = 0;
   bool  suc = taosGetSysMemory(&sysMemoryUsedMB);
   if (!suc) {
-    monitorError("monitor:%p, get sys memory info failed.", tsMonitorConn.conn);
+    monitorDebug("failed to get sys memory info");
   }
 
   float procMemoryUsedMB = 0;
   suc = taosGetProcMemory(&procMemoryUsedMB);
   if (!suc) {
-    monitorError("monitor:%p, get proc memory info failed.", tsMonitorConn.conn);
+    monitorDebug("failed to get proc memory info");
   }
 
   return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
@@ -296,11 +274,11 @@ static int32_t monitorBuildCpuSql(char *sql) {
   float sysCpuUsage = 0, procCpuUsage = 0;
   bool  suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
   if (!suc) {
-    monitorError("monitor:%p, get cpu usage failed.", tsMonitorConn.conn);
+    monitorDebug("failed to get cpu usage");
   }
 
   if (sysCpuUsage <= procCpuUsage) {
-    sysCpuUsage = procCpuUsage + (float)0.1;
+    sysCpuUsage = procCpuUsage + 0.1f;
   }
 
   return sprintf(sql, ", %f, %f, %d", procCpuUsage, sysCpuUsage, tsNumOfCores);
@@ -316,14 +294,14 @@ static int32_t monitorBuildBandSql(char *sql) {
   float bandSpeedKb = 0;
   bool  suc = taosGetBandSpeed(&bandSpeedKb);
   if (!suc) {
-    monitorError("monitor:%p, get bandwidth speed failed.", tsMonitorConn.conn);
+    monitorDebug("failed to get bandwidth speed");
   }
 
   return sprintf(sql, ", %f", bandSpeedKb);
 }
 
 static int32_t monitorBuildReqSql(char *sql) {
-  SDnodeStatisInfo info = dnodeGetStatisInfo(); 
+  SDnodeStatisInfo info = dnodeGetStatisInfo();
   return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
 }
 
@@ -331,20 +309,15 @@ static int32_t monitorBuildIoSql(char *sql) {
   float readKB = 0, writeKB = 0;
   bool  suc = taosGetProcIO(&readKB, &writeKB);
   if (!suc) {
-    monitorError("monitor:%p, get io info failed.", tsMonitorConn.conn);
+    monitorDebug("failed to get io info");
   }
 
   return sprintf(sql, ", %f, %f", readKB, writeKB);
 }
 
 static void monitorSaveSystemInfo() {
-  if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) {
-    monitorStartTimer();
-    return;
-  }
-
   int64_t ts = taosGetTimestampUs();
-  char *  sql = tsMonitorConn.sql;
+  char *  sql = tsMonitor.sql;
   int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn%d values(%" PRId64, tsMonitorDbName, dnodeGetDnodeId(), ts);
 
   pos += monitorBuildCpuSql(sql + pos);
@@ -354,16 +327,31 @@ static void monitorSaveSystemInfo() {
   pos += monitorBuildIoSql(sql + pos);
   pos += monitorBuildReqSql(sql + pos);
 
-  monitorDebug("monitor:%p, save system info, sql:%s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sys");
+  void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
+  int   code = taos_errno(res);
+  taos_free_result(res);
+
+  if (code != 0) {
+    monitorError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
+  } else {
+    monitorDebug("successfully to save system info, sql:%s", tsMonitor.sql);
+  }
+}
 
-  if (tsMonitorConn.timer != NULL && tsMonitorConn.state != MONITOR_STATE_STOPPED) {
-    monitorStartTimer();
+static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
+  int32_t c = taos_errno(result);
+  if (c != TSDB_CODE_SUCCESS) {
+    monitorError("save %s failed, reason:%s", (char *)param, tstrerror(c));
+  } else {
+    int32_t rows = taos_affected_rows(result);
+    monitorDebug("save %s succ, rows:%d", (char *)param, rows);
   }
+
+  taos_free_result(result);
 }
 
 void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
-  if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
+  if (tsMonitor.state != MON_STATE_INITED) return;
 
   char sql[1024] = {0};
   sprintf(sql,
@@ -392,19 +380,16 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
           pMon->totalConns, pMon->maxConns,
           pMon->accessState);
 
-  monitorDebug("monitor:%p, save account info, sql %s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "account");
+  monitorDebug("save account info, sql:%s", sql);
+  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info");
 }
 
 void monitorSaveLog(int32_t level, const char *const format, ...) {
-  if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
+  if (tsMonitor.state != MON_STATE_INITED) return;
 
   va_list argpointer;
   char    sql[SQL_LENGTH] = {0};
   int32_t max_length = SQL_LENGTH - 30;
-
-  if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
-
   int32_t len = snprintf(sql, (size_t)max_length, "insert into %s.log values(%" PRId64 ", %d,'", tsMonitorDbName,
                          taosGetTimestampUs(), level);
 
@@ -416,12 +401,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
   len += sprintf(sql + len, "', '%s')", tsLocalEp);
   sql[len++] = 0;
 
-  monitorDebug("monitor:%p, save log, sql: %s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "log");
+  monitorDebug("save log, sql: %s", sql);
+  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log");
 }
 
 void monitorExecuteSQL(char *sql) {
-  if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
-  monitorDebug("monitor:%p, execute sql: %s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sql");
+  if (tsMonitor.state != MON_STATE_INITED) return;
+
+  monitorDebug("execute sql:%s", sql);
+  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql");
 }

src/plugins/mqtt/CMakeLists.txt

@@ -2,21 +2,19 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
 PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/include)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/examples/templates)
 AUX_SOURCE_DIRECTORY(src SRC)
 
 IF (TD_LINUX)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/include)
-  INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/examples/templates)
   ADD_LIBRARY(mqtt ${SRC})
+  TARGET_LINK_LIBRARIES(mqtt cJson mqttc)
 
   IF (TD_SOMODE_STATIC)
-    TARGET_LINK_LIBRARIES(mqtt taos_static cJson mqttc)
+    TARGET_LINK_LIBRARIES(mqtt taos_static)
   ELSE ()
-    TARGET_LINK_LIBRARIES(mqtt taos cJson mqttc)
-  ENDIF ()
-  IF (TD_ADMIN)
-    TARGET_LINK_LIBRARIES(mqtt admin cJson)
+    TARGET_LINK_LIBRARIES(mqtt taos)
   ENDIF ()
 ENDIF ()

src/plugins/mqtt/inc/mqttInit.h

@@ -23,11 +23,12 @@ extern "C" {
  * @file
  * A simple subscriber program that performs automatic reconnections.
  */
-#include <stdio.h>
-#include <stdlib.h>
-#include <unistd.h>
 #include "mqtt.h"
-#include "taos.h"
+
+#define QOS                1
+#define TIMEOUT            10000L
+#define MQTT_SEND_BUF_SIZE 102400
+#define MQTT_RECV_BUF_SIZE 102400
 
 /**
  * @brief A structure that I will use to keep track of some data needed
@@ -36,18 +37,12 @@ extern "C" {
  * An instance of this struct will be created in my \c main(). Then, whenever
  * \ref mqttReconnectClient is called, this instance will be passed.
  */
-struct reconnect_state_t {
-  char*    hostname;
-  char*    port;
-  char*    topic;
-  char*    client_id;
-  char*    user_name;
-  char*    password;
+typedef struct SMqttReconnectState {
   uint8_t* sendbuf;
   size_t   sendbufsz;
   uint8_t* recvbuf;
   size_t   recvbufsz;
-};
+} SMqttReconnectState;
 
 /**
  * @brief My reconnect callback. It will reestablish the connection whenever
@@ -58,7 +53,7 @@ void mqttReconnectClient(struct mqtt_client* client, void** reconnect_state_vptr
 /**
  * @brief The function will be called whenever a PUBLISH message is received.
  */
-void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published);
+void mqttPublishCallback(void** unused, struct mqtt_response_publish* published);
 
 /**
  * @brief The client's refresher. This function triggers back-end routines to
@@ -73,12 +68,7 @@ void* mqttClientRefresher(void* client);
 /**
  * @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
  */
-
-void mqttCleanup(int status, int sockfd, pthread_t* client_daemon);
-void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code);
-void mqttQueryInsertCallback(void* param, TAOS_RES* result, int32_t code);
-#define QOS 1
-#define TIMEOUT 10000L
+void mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon);
 
 #ifdef __cplusplus
 }

src/plugins/mqtt/inc/mqttPayload.h

@@ -15,11 +15,13 @@
 
 #ifndef TDENGINE_MQTT_PLYLOAD_H
 #define TDENGINE_MQTT_PLYLOAD_H
+
 #ifdef __cplusplus
 extern "C" {
 #endif
-char  split(char str[], char delims[], char** p_p_cmd_part, int max);
-char* converJsonToSql(char* json, char* _dbname, char* _tablename);
+
+char* mqttConverJsonToSql(char* json, int maxSize);
+
 #ifdef __cplusplus
 }
 #endif

src/plugins/mqtt/src/mqttPayload.c

@@ -14,52 +14,146 @@
  */
 
 #define _DEFAULT_SOURCE
-#include "mqttPayload.h"
+#include "os.h"
 #include "cJSON.h"
-#include "string.h"
-#include "taos.h"
 #include "mqttLog.h"
-#include "os.h"
-char split(char str[], char delims[], char** p_p_cmd_part, int max) {
-  char*  token = strtok(str, delims);
-  char   part_index = 0;
-  char** tmp_part = p_p_cmd_part;
-  while (token) {
-    *tmp_part++ = token;
-    token = strtok(NULL, delims);
-    part_index++;
-    if (part_index >= max) break;
-  }
-  return part_index;
+#include "mqttPayload.h"
+
+// subscribe message like this
+
+/*
+/test {
+    "timestamp": 1599121290,
+    "gateway": {
+        "name": "AcuLink 810 Gateway",
+        "model": "AcuLink810-868",
+        "serial": "S8P20200207"
+    },
+    "device": {
+        "name": "Acuvim L V3 .221",
+        "model": "Acuvim-L-V3",
+        "serial": "221",
+        "online": true,
+        "readings": [
+            {
+                "param": "Freq_Hz",
+                "value": "59.977539",
+                "unit": "Hz"
+            },
+            {
+                "param": "Va_V",
+                "value": "122.002907",
+                "unit": "V"
+            },
+            {
+                "param": "DI4",
+                "value": "5.000000",
+                "unit": ""
+            }
+        ]
+    }
 }
+*/
+
+// send msg cmd
+// mosquitto_pub -h test.mosquitto.org  -t "/test" -m '{"timestamp": 1599121290,"gateway": {"name": "AcuLink 810 Gateway","model": "AcuLink810-868","serial": "S8P20200207"},"device": {"name": "Acuvim L V3 .221","model": "Acuvim-L-V3","serial": "221","online": true,"readings": [{"param": "Freq_Hz","value": "59.977539","unit": "Hz"},{"param": "Va_V","value": "122.002907","unit": "V"},{"param": "DI4","value": "5.000000","unit": ""}]}}'
+
+/*
+ * This is an example, this function needs to be implemented in order to parse the json file into a sql statement
+ * Note that you need to create a super table and database before writing data
+ * In this case:
+ *   create database mqttdb;
+ *   create table mqttdb.devices(ts timestamp, value bigint) tags(name binary(32), model binary(32), serial binary(16), param binary(16), unit binary(16));
+ */
+
+char* mqttConverJsonToSql(char* json, int maxSize) {
+  // const int32_t maxSize = 10240;
+  maxSize *= 5;
+  char* sql = malloc(maxSize);
+
+  cJSON* root = cJSON_Parse(json);
+  if (root == NULL) {
+    mqttError("failed to parse msg, invalid json format");
+    goto MQTT_PARSE_OVER;
+  }
+
+  cJSON* timestamp = cJSON_GetObjectItem(root, "timestamp");
+  if (!timestamp || timestamp->type != cJSON_Number) {
+    mqttError("failed to parse msg, timestamp not found");
+    goto MQTT_PARSE_OVER;
+  }
+
+  cJSON* device = cJSON_GetObjectItem(root, "device");
+  if (!device) {
+    mqttError("failed to parse msg, device not found");
+    goto MQTT_PARSE_OVER;
+  }
 
-char* converJsonToSql(char* json, char* _dbname, char* _tablename) {
-  cJSON* jPlayload = cJSON_Parse(json);
-  char   _names[102400] = {0};
-  char   _values[102400] = {0};
-  int    i = 0;
-  int    count = cJSON_GetArraySize(jPlayload);
-  for (; i < count; i++)  
-  {
-    cJSON* item = cJSON_GetArrayItem(jPlayload, i);
-    if (cJSON_Object == item->type) {
-      mqttInfo("The item '%s' is not supported", item->string);
-    } else {
-      strcat(_names, item->string);
-      if (i < count - 1) {
-        strcat(_names, ",");
-      }
-      char* __value_json = cJSON_Print(item);
-      strcat(_values, __value_json);
-      free(__value_json);
-      if (i < count - 1) {
-        strcat(_values, ",");
-      }
+  cJSON* name = cJSON_GetObjectItem(device, "name");
+  if (!name || name->type != cJSON_String) {
+    mqttError("failed to parse msg, name not found");
+    goto MQTT_PARSE_OVER;
+  }
+
+  cJSON* model = cJSON_GetObjectItem(device, "model");
+  if (!model || model->type != cJSON_String) {
+    mqttError("failed to parse msg, model not found");
+    goto MQTT_PARSE_OVER;
+  }
+
+  cJSON* serial = cJSON_GetObjectItem(device, "serial");
+  if (!serial || serial->type != cJSON_String) {
+    mqttError("failed to parse msg, serial not found");
+    goto MQTT_PARSE_OVER;
+  }
+
+  cJSON* readings = cJSON_GetObjectItem(device, "readings");
+  if (!readings || readings->type != cJSON_Array) {
+    mqttError("failed to parse msg, readings not found");
+    goto MQTT_PARSE_OVER;
+  }
+
+  int count = cJSON_GetArraySize(readings);
+  if (count <= 0) {
+    mqttError("failed to parse msg, readings size smaller than 0");
+    goto MQTT_PARSE_OVER;
+  }
+
+  int len = snprintf(sql, maxSize, "insert into");
+
+  for (int i = 0; i < count; ++i) {
+    cJSON* reading = cJSON_GetArrayItem(readings, i);
+    if (reading == NULL) continue;
+
+    cJSON* param = cJSON_GetObjectItem(reading, "param");
+    if (!param || param->type != cJSON_String) {
+      mqttError("failed to parse msg, param not found");
+      goto MQTT_PARSE_OVER;
     }
+
+    cJSON* value = cJSON_GetObjectItem(reading, "value");
+    if (!value || value->type != cJSON_String) {
+      mqttError("failed to parse msg, value not found");
+      goto MQTT_PARSE_OVER;
+    }
+
+    cJSON* unit = cJSON_GetObjectItem(reading, "unit");
+    if (!unit || unit->type != cJSON_String) {
+      mqttError("failed to parse msg, unit not found");
+      goto MQTT_PARSE_OVER;
+    }
+
+    len += snprintf(sql + len, maxSize - len,
+        " mqttdb.serial_%s_%s using mqttdb.devices tags('%s', '%s', '%s', '%s', '%s') values(%" PRId64 ", %s)",
+        serial->valuestring, param->valuestring, name->valuestring, model->valuestring, serial->valuestring,
+        param->valuestring, unit->valuestring, timestamp->valueint * 1000, value->valuestring);
   }
-  cJSON_free(jPlayload);
-  int   sqllen = strlen(_names) + strlen(_values) + strlen(_dbname) + strlen(_tablename) + 1024;
-  char* _sql = calloc(1, sqllen);
-  sprintf(_sql, "INSERT INTO %s.%s (%s) VALUES(%s);", _dbname, _tablename, _names, _values);
-  return _sql;
+
+  cJSON_free(root);
+  return sql;
+
+MQTT_PARSE_OVER:
+  cJSON_free(root);
+  free(sql);
+  return NULL;
 }
\ No newline at end of file

src/plugins/mqtt/src/mqttSystem.c

@@ -14,244 +14,131 @@
  */
 
 #define _DEFAULT_SOURCE
-
-#include "cJSON.h"
+#include "os.h"
 #include "mqtt.h"
 #include "mqttInit.h"
 #include "mqttLog.h"
 #include "mqttPayload.h"
-#include "os.h"
+#include "tmqtt.h"
 #include "posix_sockets.h"
-#include "string.h"
 #include "taos.h"
 #include "tglobal.h"
-#include "tmqtt.h"
-#include "tsclient.h"
-#include "tsocket.h"
-#include "ttimer.h"
-#include "mqttSystem.h"
-struct mqtt_client       mqttClient = {0};
-pthread_t                clientDaemonThread = {0};
-void*                    mqttConnect=NULL;
-struct reconnect_state_t recntStatus = {0};
-char*                    topicPath=NULL;
-int                      mttIsRuning = 1;
+#include "taoserror.h"
 
-int32_t mqttInitSystem() {
-  int   rc = 0;
-#if 0
-  uint8_t sendbuf[2048];
-  uint8_t recvbuf[1024];
-  recntStatus.sendbuf = sendbuf;
-  recntStatus.sendbufsz = sizeof(sendbuf);
-  recntStatus.recvbuf = recvbuf;
-  recntStatus.recvbufsz = sizeof(recvbuf);
-  char* url = tsMqttBrokerAddress;
-  recntStatus.user_name = strstr(url, "@") != NULL ? strbetween(url, "//", ":") : NULL;
-  
-  char * passStr = strstr(url, recntStatus.user_name);
-  if (passStr != NULL) {
-    recntStatus.password = strstr(url, "@") != NULL ? strbetween(passStr, ":", "@") : NULL;
-  }
+struct SMqttReconnectState tsMqttStatus = {0};
+struct mqtt_client tsMqttClient = {0};
+static pthread_t   tsMqttClientDaemonThread = {0};
+static void*       tsMqttConnect = NULL;
+static bool        tsMqttIsRuning = false;
 
-  if (strlen(url) == 0) {
-    mqttDebug("mqtt module not init, url is null");
-    return rc;
-  }
-
-  if (strstr(url, "@") != NULL) {
-    recntStatus.hostname = strbetween(url, "@", ":");
-  } else if (strstr(strstr(url, "://") + 3, ":") != NULL) {
-    recntStatus.hostname = strbetween(url, "//", ":");
-
-  } else {
-    recntStatus.hostname = strbetween(url, "//", "/");
-  }
+int32_t mqttInitSystem() { return 0; }
 
-  char* _begin_hostname = strstr(url, recntStatus.hostname);
-  if (_begin_hostname != NULL && strstr(_begin_hostname, ":") != NULL) {
-    recntStatus.port = strbetween(_begin_hostname, ":", "/");
-  } else {
-    recntStatus.port = strbetween("'1883'", "'", "'");
-  }
+int32_t mqttStartSystem() {
+  tsMqttStatus.sendbufsz = MQTT_SEND_BUF_SIZE;
+  tsMqttStatus.recvbufsz = MQTT_RECV_BUF_SIZE;
+  tsMqttStatus.sendbuf = malloc(MQTT_SEND_BUF_SIZE);
+  tsMqttStatus.recvbuf = malloc(MQTT_RECV_BUF_SIZE);
+  tsMqttIsRuning = true;
 
-  char* portStr = recntStatus.hostname;
-  if (_begin_hostname != NULL) {
-    char* colonStr = strstr(_begin_hostname, ":");
-    if (colonStr != NULL) {
-      portStr = recntStatus.port;
-    }
+  mqtt_init_reconnect(&tsMqttClient, mqttReconnectClient, &tsMqttStatus, mqttPublishCallback);
+  if (pthread_create(&tsMqttClientDaemonThread, NULL, mqttClientRefresher, &tsMqttClient)) {
+    mqttError("mqtt failed to start daemon.");
+    mqttCleanupRes(EXIT_FAILURE, -1, NULL);
+    return -1;
   }
 
-  char* topicStr = strstr(url, portStr);
-  if (topicStr != NULL) {
-    topicPath = strbetween(topicStr, "/", "/");
-    char* _topic = "+/+/+/";
-    int   _tpsize = strlen(topicPath) + strlen(_topic) + 1;
-    recntStatus.topic = calloc(1, _tpsize);
-    sprintf(recntStatus.topic, "/%s/%s", topicPath, _topic);
-    recntStatus.client_id = strlen(tsMqttBrokerClientId) < 3 ? tsMqttBrokerClientId : "taos_mqtt";
-    mqttConnect = NULL;
-  } else {
-    topicPath = NULL;
-  }
-  
-#endif  
-  return rc;
+  mqttInfo("mqtt listening for topic:%s messages", tsMqttTopic);
+  return 0;
 }
 
-int32_t mqttStartSystem() {
-  int rc = 0;
-#if 0  
-  if (recntStatus.user_name != NULL && recntStatus.password != NULL) {
-    mqttInfo("connecting to  mqtt://%s:%s@%s:%s/%s/", recntStatus.user_name, recntStatus.password,
-              recntStatus.hostname, recntStatus.port, topicPath);
-  } else if (recntStatus.user_name != NULL && recntStatus.password == NULL) {
-    mqttInfo("connecting to  mqtt://%s@%s:%s/%s/", recntStatus.user_name, recntStatus.hostname, recntStatus.port,
-              topicPath);
-  }
+void mqttStopSystem() {
+  if (tsMqttIsRuning) {
+    tsMqttIsRuning = false;
+    tsMqttClient.error = MQTT_ERROR_SOCKET_ERROR;
 
-  mqtt_init_reconnect(&mqttClient, mqttReconnectClient, &recntStatus, mqtt_PublishCallback);
-  if (pthread_create(&clientDaemonThread, NULL, mqttClientRefresher, &mqttClient)) {
-    mqttError("Failed to start client daemon.");
-    mqttCleanup(EXIT_FAILURE, -1, NULL);
-    rc = -1;
-  } else {
-    mqttInfo("listening for '%s' messages.", recntStatus.topic);
-  }
-#endif
-  return rc;
-}
+    taosMsleep(300);
+    mqttCleanupRes(EXIT_SUCCESS, tsMqttClient.socketfd, &tsMqttClientDaemonThread);
 
-void mqttStopSystem() {
-#if 0  
-  mqttClient.error = MQTT_ERROR_SOCKET_ERROR;
-  mttIsRuning = 0;
-  usleep(300000U);
-  mqttCleanup(EXIT_SUCCESS, mqttClient.socketfd, &clientDaemonThread);
-  mqttInfo("mqtt is stoped");
-#endif  
+    mqttInfo("mqtt is stopped");
+  }
 }
 
 void mqttCleanUpSystem() {
-#if 0
-  mqttInfo("starting to cleanup mqtt");
-  free(recntStatus.user_name);
-  free(recntStatus.password);
-  free(recntStatus.hostname);
-  free(recntStatus.port);
-  free(recntStatus.topic);
-  free(topicPath);
+  mqttStopSystem();
   mqttInfo("mqtt is cleaned up");
-#endif  
 }
 
-void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published) {
-  /* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
-  char* topic_name = (char*)malloc(published->topic_name_size + 1);
-  memcpy(topic_name, published->topic_name, published->topic_name_size);
-  topic_name[published->topic_name_size] = '\0';
-  mqttInfo("received publish('%s'): %s", topic_name, (const char*)published->application_message);
-  char _token[128] = {0};
-  char _dbname[128] = {0};
-  char _tablename[128] = {0};
-  if (mqttConnect == NULL) {
-    mqttInfo("connect database");
-    taos_connect_a(NULL, "_root", tsInternalPass, "", 0, mqttInitConnCb, &mqttClient, &mqttConnect);
-  }
-  if (topic_name[1]=='/' &&  strncmp((char*)&topic_name[1], topicPath, strlen(topicPath)) == 0) {
-    char* p_p_cmd_part[5] = {0};
-    char  copystr[1024] = {0};
-    strncpy(copystr, topic_name, MIN(1024, published->topic_name_size));
-    char part_index = split(copystr, "/", p_p_cmd_part, 10);
-    if (part_index < 4) {
-      mqttError("The topic %s is't format '/path/token/dbname/table name/'. for expmle: '/taos/token/db/t'", topic_name);
+void mqttPublishCallback(void** unused, struct mqtt_response_publish* published) {
+  const char* content = published->application_message;
+  mqttDebug("receive mqtt message, size:%d", (int)published->application_message_size);
+
+  if (tsMqttConnect == NULL) {
+    tsMqttConnect = taos_connect(NULL, "_root", tsInternalPass, "", 0);
+    if (tsMqttConnect == NULL) {
+      mqttError("failed to connect to tdengine, reason:%s", tstrerror(terrno));
+      return;
     } else {
-      strncpy(_token, p_p_cmd_part[1], 127);
-      strncpy(_dbname, p_p_cmd_part[2], 127);
-      strncpy(_tablename, p_p_cmd_part[3], 127);
-      mqttInfo("part count=%d,access token:%s,database name:%s, table name:%s", part_index, _token, _dbname,
-                _tablename);
+      mqttInfo("successfully connected to the tdengine");
+    }
+  }
 
-      if (mqttConnect != NULL) {
-        char* _sql = converJsonToSql((char*)published->application_message, _dbname, _tablename);
-        mqttInfo("query:%s", _sql);
-        taos_query_a(mqttConnect, _sql, mqttQueryInsertCallback, &mqttClient);
-        mqttInfo("free sql:%s", _sql);
-        free(_sql);
-      }
+  mqttTrace("receive mqtt message, content:%s", content);
+
+  char* sql = mqttConverJsonToSql((char*)content, (int)published->application_message_size);
+  if (sql != NULL) {
+    void* res = taos_query(tsMqttConnect, sql);
+    int   code = taos_errno(res);
+    if (code != 0) {
+      mqttError("failed to exec sql, reason:%s sql:%s", tstrerror(code), sql);
+    } else {
+      mqttTrace("successfully to exec sql:%s", sql);
     }
+    taos_free_result(res);
+  } else {
+    mqttError("failed to parse mqtt message");
   }
-  free(topic_name);
 }
 
 void* mqttClientRefresher(void* client) {
-  while (mttIsRuning) {
+  while (tsMqttIsRuning) {
     mqtt_sync((struct mqtt_client*)client);
     taosMsleep(100);
   }
-  mqttDebug("quit refresher");
+
+  mqttDebug("mqtt quit refresher");
   return NULL;
 }
 
-void mqttCleanup(int status, int sockfd, pthread_t* client_daemon) {
-#if 0 
+void mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon) {
   mqttInfo("clean up mqtt module");
-  if (sockfd != -1) close(sockfd);
-  if (client_daemon != NULL) pthread_cancel(*client_daemon);
-#endif
-}
-
-void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code) {
-  if (code < 0) {
-    mqttError("mqtt:%d, connect to database failed, reason:%s", code, tstrerror(code));
-    taos_close(mqttConnect);
-    mqttConnect = NULL;
-    return;
+  if (sockfd != -1) {
+    close(sockfd);
   }
-  mqttDebug("mqtt:%d, connect to database success, reason:%s", code, tstrerror(code));
-}
 
-void mqttQueryInsertCallback(void* param, TAOS_RES* result, int32_t code) {
-  if (code < 0) {
-    mqttError("mqtt:%d, save data failed, code:%s", code, tstrerror(code));
-  } else if (code == 0) {
-    mqttError("mqtt:%d, save data failed, affect rows:%d", code, code);
-  } else {
-    mqttInfo("mqtt:%d, save data   success, code:%s", code, tstrerror(code));
+  if (client_daemon != NULL) {
+    pthread_cancel(*client_daemon);
   }
 }
 
-void mqttReconnectClient(struct mqtt_client* client, void** reconnect_state_vptr) {
-  mqttInfo("reconnect client");
-  struct reconnect_state_t* reconnect_state = *((struct reconnect_state_t**)reconnect_state_vptr);
+void mqttReconnectClient(struct mqtt_client* client, void** unused) {
+  mqttInfo("mqtt tries to connect to the mqtt server");
 
-  /* Close the clients socket if this isn't the initial reconnect call */
   if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
     close(client->socketfd);
   }
 
-  /* Perform error handling here. */
   if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
-    mqttError("mqttReconnectClient: called while client was in error state \"%s\"", mqtt_error_str(client->error));
+    mqttError("mqtt client was in error state %s", mqtt_error_str(client->error));
   }
 
-  /* Open a new socket. */
-  int sockfd = open_nb_socket(reconnect_state->hostname, reconnect_state->port);
-  if (sockfd == -1) {
-    mqttError("failed to open socket: ");
-    mqttCleanup(EXIT_FAILURE, sockfd, NULL);
+  int sockfd = open_nb_socket(tsMqttHostName, tsMqttPort);
+  if (sockfd < 0) {
+    mqttError("mqtt client failed to open socket %s:%s", tsMqttHostName, tsMqttPort);
+    //mqttCleanupRes(EXIT_FAILURE, sockfd, NULL);
+    return;
   }
 
-  /* Reinitialize the client. */
-  mqtt_reinit(client, sockfd, reconnect_state->sendbuf, reconnect_state->sendbufsz, reconnect_state->recvbuf,
-              reconnect_state->recvbufsz);
-
-  /* Ensure we have a clean session */
-  uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
-  /* Send connection request to the broker. */
-  mqtt_connect(client, reconnect_state->client_id, NULL, NULL, 0, reconnect_state->user_name, reconnect_state->password,connect_flags, 400);
-
-  /* Subscribe to the topic. */
-  mqtt_subscribe(client, reconnect_state->topic, 0);
+  mqtt_reinit(client, sockfd, tsMqttStatus.sendbuf, tsMqttStatus.sendbufsz, tsMqttStatus.recvbuf, tsMqttStatus.recvbufsz);
+  mqtt_connect(client, tsMqttClientId, NULL, NULL, 0, tsMqttUser, tsMqttPass, MQTT_CONNECT_CLEAN_SESSION, 400);
+  mqtt_subscribe(client, tsMqttTopic, 0);
 }
\ No newline at end of file

src/query/CMakeLists.txt

@@ -8,10 +8,9 @@ INCLUDE_DIRECTORIES(inc)
 AUX_SOURCE_DIRECTORY(src SRC)
 ADD_LIBRARY(query ${SRC})
 SET_SOURCE_FILES_PROPERTIES(src/sql.c PROPERTIES COMPILE_FLAGS -w)
+TARGET_LINK_LIBRARIES(query tsdb tutil)
 
 IF (TD_LINUX)
-  TARGET_LINK_LIBRARIES(query tsdb tutil m rt)
+  TARGET_LINK_LIBRARIES(query m rt)
   ADD_SUBDIRECTORY(tests)
-ELSEIF (TD_WINDOWS)
-   TARGET_LINK_LIBRARIES(query tsdb tutil)
 ENDIF ()

src/query/inc/sql.y

@@ -217,7 +217,6 @@ acct_optr(Y) ::= pps(C) tseries(D) storage(P) streams(F) qtime(Q) dbs(E) users(K
 %destructor keep {tVariantListDestroy($$);}
 keep(Y)    ::= KEEP tagitemlist(X).           { Y = X; }
 
-tables(Y)  ::= MAXTABLES INTEGER(X).          { Y = X; }
 cache(Y)   ::= CACHE INTEGER(X).              { Y = X; }
 replica(Y) ::= REPLICA INTEGER(X).            { Y = X; }
 quorum(Y)  ::= QUORUM INTEGER(X).             { Y = X; }
@@ -234,7 +233,6 @@ prec(Y)    ::= PRECISION STRING(X).           { Y = X; }
 %type db_optr {SCreateDBInfo}
 db_optr(Y) ::= . {setDefaultCreateDbOption(&Y);}
 
-db_optr(Y) ::= db_optr(Z) tables(X).         { Y = Z; Y.maxTablesPerVnode = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) cache(X).          { Y = Z; Y.cacheBlockSize = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) replica(X).        { Y = Z; Y.replica = strtol(X.z, NULL, 10); }
 db_optr(Y) ::= db_optr(Z) quorum(X).         { Y = Z; Y.quorum = strtol(X.z, NULL, 10); }
@@ -254,7 +252,6 @@ alter_db_optr(Y) ::= . { setDefaultCreateDbOption(&Y);}
 
 alter_db_optr(Y) ::= alter_db_optr(Z) replica(X).     { Y = Z; Y.replica = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) quorum(X).      { Y = Z; Y.quorum = strtol(X.z, NULL, 10); }
-alter_db_optr(Y) ::= alter_db_optr(Z) tables(X).      { Y = Z; Y.maxTablesPerVnode = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) keep(X).        { Y = Z; Y.keep = X; }
 alter_db_optr(Y) ::= alter_db_optr(Z) blocks(X).      { Y = Z; Y.numOfBlocks = strtol(X.z, NULL, 10); }
 alter_db_optr(Y) ::= alter_db_optr(Z) comp(X).        { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }

src/query/src/qTokenizer.c

@@ -121,7 +121,6 @@ static SKeyword keywordTable[] = {
     {"MINROWS",      TK_MINROWS},
     {"MAXROWS",      TK_MAXROWS},
     {"BLOCKS",       TK_BLOCKS},
-    {"MAXTABLES",    TK_MAXTABLES},
     {"CACHE",        TK_CACHE},
     {"CTIME",        TK_CTIME},
     {"WAL",          TK_WAL},

src/rpc/src/rpcMain.c

@@ -709,21 +709,21 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
       }
 
       if (terrno != 0) {
-        taosFreeId(pRpc->idPool, sid);   // sid shall be released
+        taosFreeId(pRpc->idPool, sid);  // sid shall be released
         pConn = NULL;
       }
     }
-  }      
+  }
 
   if (pConn) {
     if (pRecv->connType == RPC_CONN_UDPS && pRpc->numOfThreads > 1) {
       // UDP server, assign to new connection
-      pRpc->index = (pRpc->index+1) % pRpc->numOfThreads;
+      pRpc->index = (pRpc->index + 1) % pRpc->numOfThreads;
       pConn->localPort = (pRpc->localPort + pRpc->index);
     }
-  
+
     taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES);
-    tDebug("%s %p server connection is allocated, uid:0x%x", pRpc->label, pConn, pConn->linkUid);
+    tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s", pRpc->label, pConn, pConn->linkUid, sid, hashstr);
   }
 
   return pConn;

src/tsdb/CMakeLists.txt

@@ -3,13 +3,10 @@ PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(inc)
 AUX_SOURCE_DIRECTORY(src SRC)
+ADD_LIBRARY(tsdb ${SRC})
+TARGET_LINK_LIBRARIES(tsdb common tutil)
 
 IF (TD_LINUX)
-  ADD_LIBRARY(tsdb ${SRC})
-  TARGET_LINK_LIBRARIES(tsdb common tutil)
   # Someone has no gtest directory, so comment it
   # ADD_SUBDIRECTORY(tests)
-ELSEIF (TD_WINDOWS)
-  ADD_LIBRARY(tsdb ${SRC})
-  TARGET_LINK_LIBRARIES(tsdb common tutil)
 ENDIF ()

src/tsdb/inc/tsdbMain.h

@@ -45,6 +45,8 @@ extern int tsdbDebugFlag;
 #define TSDB_FILE_DELIMITER 0xF00AFA0F
 #define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
 
+#define TAOS_IN_RANGE(key, keyMin, keyLast) (((key) >= (keyMin)) && ((key) <= (keyMax)))
+
 // NOTE: Any file format change must increase this version number by 1
 //       Also, implement the convert function
 #define TSDB_FILE_VERSION ((uint32_t)0)
@@ -318,6 +320,16 @@ typedef struct {
   void*      compBuffer;  // Buffer for temperary compress/decompress purpose
 } SRWHelper;
 
+// ------------------ tsdbScan.c
+typedef struct {
+  SFileGroup fGroup;
+  int        numOfIdx;
+  SCompIdx*  pCompIdx;
+  SCompInfo* pCompInfo;
+  void*      pBuf;
+  FILE*      tLogStream;
+} STsdbScanHandle;
+
 // Operations
 // ------------------ tsdbMeta.c
 #define TSDB_INIT_NTABLES 1024
@@ -475,6 +487,7 @@ int         tsdbUpdateFileHeader(SFile* pFile);
 int         tsdbEncodeSFileInfo(void** buf, const STsdbFileInfo* pInfo);
 void*       tsdbDecodeSFileInfo(void* buf, STsdbFileInfo* pInfo);
 void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
+int         tsdbLoadFileHeader(SFile* pFile, uint32_t* version);
 void        tsdbGetFileInfoImpl(char* fname, uint32_t* magic, int64_t* size);
 void        tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
 
@@ -513,7 +526,10 @@ int  tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols
 int  tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
 int  tsdbWriteCompInfo(SRWHelper* pHelper);
 int  tsdbWriteCompIdx(SRWHelper* pHelper);
+int  tsdbLoadCompIdxImpl(SFile* pFile, uint32_t offset, uint32_t len, void* buffer);
+int  tsdbDecodeSCompIdxImpl(void* buffer, uint32_t len, SCompIdx** ppCompIdx, int* numOfIdx);
 int  tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
+int  tsdbLoadCompInfoImpl(SFile* pFile, SCompIdx* pIdx, SCompInfo** ppCompInfo);
 int  tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
 int  tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
 void tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
@@ -537,7 +553,7 @@ static FORCE_INLINE int compTSKEY(const void* key1, const void* key2) {
 #define TSDB_SUBMIT_MSG_HEAD_SIZE sizeof(SSubmitMsg)
 
 char*       tsdbGetMetaFileName(char* rootDir);
-void        tsdbGetDataFileName(STsdbRepo* pRepo, int fid, int type, char* fname);
+void        tsdbGetDataFileName(char* rootDir, int vid, int fid, int type, char* fname);
 int         tsdbLockRepo(STsdbRepo* pRepo);
 int         tsdbUnlockRepo(STsdbRepo* pRepo);
 char*       tsdbGetDataDirName(char* rootDir);
@@ -546,6 +562,16 @@ STsdbMeta*  tsdbGetMeta(TSDB_REPO_T* pRepo);
 STsdbFileH* tsdbGetFile(TSDB_REPO_T* pRepo);
 int         tsdbCheckCommit(STsdbRepo* pRepo);
 
+// ------------------ tsdbScan.c
+int              tsdbScanFGroup(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
+STsdbScanHandle* tsdbNewScanHandle();
+void             tsdbSetScanLogStream(STsdbScanHandle* pScanHandle, FILE* fLogStream);
+int              tsdbSetAndOpenScanFile(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
+int              tsdbScanSCompIdx(STsdbScanHandle* pScanHandle);
+int              tsdbScanSCompBlock(STsdbScanHandle* pScanHandle, int idx);
+int              tsdbCloseScanFile(STsdbScanHandle* pScanHandle);
+void             tsdbFreeScanHandle(STsdbScanHandle* pScanHandle);
+
 #ifdef __cplusplus
 }
 #endif

src/tsdb/src/tsdbRecover.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
\ No newline at end of file