Merge branch 'develop' into coverity_scan

.travis.yml

@@ -63,7 +63,7 @@ matrix:
           pkill -TERM -x taosd
           fuser -k -n tcp 6030
           sleep 1
-          ./crash_gen.sh -a -p -t 4 -s 25|| travis_terminate $?
+          ./crash_gen.sh -a -p -t 4 -s 2000|| travis_terminate $?
           sleep 1
 
           cd ${TRAVIS_BUILD_DIR}/tests/pytest

src/client/inc/tscLocalMerge.h

@@ -20,8 +20,8 @@
 extern "C" {
 #endif
 
-#include "qextbuffer.h"
-#include "qfill.h"
+#include "qExtbuffer.h"
+#include "qFill.h"
 #include "taosmsg.h"
 #include "tlosertree.h"
 #include "tsclient.h"

src/client/inc/tscUtil.h

@@ -23,11 +23,11 @@ extern "C" {
 /*
  * @date   2018/09/30
  */
-#include "os.h"
-#include "tbuffer.h"
 #include "exception.h"
-#include "qextbuffer.h"
+#include "os.h"
+#include "qExtbuffer.h"
 #include "taosdef.h"
+#include "tbuffer.h"
 #include "tscLocalMerge.h"
 #include "tsclient.h"
 

src/client/inc/tsclient.h

@@ -31,8 +31,8 @@ extern "C" {
 #include "tutil.h"
 
 #include "qExecutor.h"
+#include "qTsbuf.h"
 #include "qsqlparser.h"
-#include "qtsbuf.h"
 #include "tcmdtype.h"
 
 // forward declaration

src/client/src/tscFunctionImpl.c

@@ -14,15 +14,15 @@
  */
 
 #include "os.h"
-#include "qextbuffer.h"
-#include "qfill.h"
-#include "qhistogram.h"
-#include "qpercentile.h"
-#include "qsyntaxtreefunction.h"
-#include "qtsbuf.h"
+#include "qAst.h"
+#include "qExtbuffer.h"
+#include "qFill.h"
+#include "qHistogram.h"
+#include "qPercentile.h"
+#include "qSyntaxtreefunction.h"
+#include "qTsbuf.h"
 #include "taosdef.h"
 #include "taosmsg.h"
-#include "qast.h"
 #include "tscLog.h"
 #include "tscSubquery.h"
 #include "tscompression.h"
@@ -74,7 +74,7 @@ for (int32_t i = 0; i < (ctx)->tagInfo.numOfTagCols; ++i) {                  \
 void noop1(SQLFunctionCtx *UNUSED_PARAM(pCtx)) {}
 void noop2(SQLFunctionCtx *UNUSED_PARAM(pCtx), int32_t UNUSED_PARAM(index)) {}
 
-void doFinalizer(SQLFunctionCtx *pCtx) { resetResultInfo(GET_RES_INFO(pCtx)); }
+void doFinalizer(SQLFunctionCtx *pCtx) { RESET_RESULT_INFO(GET_RES_INFO(pCtx)); }
 
 typedef struct tValuePair {
   tVariant v;
@@ -330,12 +330,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
   return TSDB_CODE_SUCCESS;
 }
 
-/**
- * the numOfRes should be kept, since it may be used later
- * and allow the ResultInfo to be re initialized
- */
-void resetResultInfo(SResultInfo *pResInfo) { pResInfo->initialized = false; }
-
 void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf) {
   assert(pResInfo->interResultBuf == NULL);
   

src/client/src/tscLocal.c

@@ -16,14 +16,14 @@
 #include "os.h"
 #include "taosmsg.h"
 
-#include "tcache.h"
-#include "tscUtil.h"
-#include "tsclient.h"
+#include "qExtbuffer.h"
 #include "taosdef.h"
+#include "tcache.h"
+#include "tname.h"
 #include "tscLog.h"
-#include "qextbuffer.h"
+#include "tscUtil.h"
 #include "tschemautil.h"
-#include "tname.h"
+#include "tsclient.h"
 
 static void tscSetLocalQueryResult(SSqlObj *pSql, const char *val, const char *columnName, int16_t type, size_t valueLength);
 

src/client/src/tscSQLParser.c

@@ -18,9 +18,9 @@
 #define _DEFAULT_SOURCE
 
 #include "os.h"
+#include "qAst.h"
 #include "taos.h"
 #include "taosmsg.h"
-#include "qast.h"
 #include "tcompare.h"
 #include "tname.h"
 #include "tscLog.h"
@@ -358,7 +358,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
     }
 
     case TSDB_SQL_CFG_DNODE: {
-      const char* msg2 = "invalid configure options or values";
+      const char* msg2 = "invalid configure options or values, such as resetlog / debugFlag 135 / balance 'vnode:1-dnode:2' / monitor 1 ";
       const char* msg3 = "invalid dnode ep";
 
       /* validate the ip address */
@@ -4674,26 +4674,42 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
     return TSDB_CODE_TSC_INVALID_SQL;
   }
 
-  const int DNODE_DYNAMIC_CFG_OPTIONS_SIZE = 19;
-  const SDNodeDynConfOption DNODE_DYNAMIC_CFG_OPTIONS[] = {
-      {"resetLog", 8},       {"resetQueryCache", 15},  {"debugFlag", 9},     {"mDebugFlag", 10},
-      {"dDebugFlag", 10},    {"sdbDebugFlag", 12},     {"vDebugFlag", 10},   {"cDebugFlag", 10},
-      {"httpDebugFlag", 13}, {"monitorDebugFlag", 16}, {"rpcDebugFlag", 12}, {"uDebugFlag", 10},
-      {"tmrDebugFlag", 12},  {"qDebugflag", 10},       {"sDebugflag", 10},   {"tsdbDebugFlag", 13},
-      {"mqttDebugFlag", 13}, {"wDebugFlag", 10},       {"monitor", 7}};
+  const int tokenLogEnd = 2;
+  const int tokenBalance = 2;
+  const int tokenMonitor = 3;
+  const int tokenDebugFlag = 4;
+  const int tokenDebugFlagEnd = 20;
+  const SDNodeDynConfOption cfgOptions[] = {
+      {"resetLog", 8},    {"resetQueryCache", 15},  {"balance", 7},     {"monitor", 7},
+      {"debugFlag", 9},   {"monitorDebugFlag", 16}, {"vDebugFlag", 10}, {"mDebugFlag", 10},
+      {"cDebugFlag", 10}, {"httpDebugFlag", 13},    {"qDebugflag", 10}, {"sdbDebugFlag", 12},
+      {"uDebugFlag", 10}, {"tsdbDebugFlag", 13},    {"sDebugflag", 10}, {"rpcDebugFlag", 12},
+      {"dDebugFlag", 10}, {"mqttDebugFlag", 13},    {"wDebugFlag", 10}, {"tmrDebugFlag", 12},
+  };
 
   SSQLToken* pOptionToken = &pOptions->a[1];
 
   if (pOptions->nTokens == 2) {
     // reset log and reset query cache does not need value
-    for (int32_t i = 0; i < 2; ++i) {
-      const SDNodeDynConfOption* pOption = &DNODE_DYNAMIC_CFG_OPTIONS[i];
+    for (int32_t i = 0; i < tokenLogEnd; ++i) {
+      const SDNodeDynConfOption* pOption = &cfgOptions[i];
       if ((strncasecmp(pOption->name, pOptionToken->z, pOptionToken->n) == 0) && (pOption->len == pOptionToken->n)) {
         return TSDB_CODE_SUCCESS;
       }
     }
-  } else if ((strncasecmp(DNODE_DYNAMIC_CFG_OPTIONS[DNODE_DYNAMIC_CFG_OPTIONS_SIZE - 1].name, pOptionToken->z, pOptionToken->n) == 0) &&
-             (DNODE_DYNAMIC_CFG_OPTIONS[DNODE_DYNAMIC_CFG_OPTIONS_SIZE - 1].len == pOptionToken->n)) {
+  } else if ((strncasecmp(cfgOptions[tokenBalance].name, pOptionToken->z, pOptionToken->n) == 0) &&
+             (cfgOptions[tokenBalance].len == pOptionToken->n)) {
+    SSQLToken* pValToken = &pOptions->a[2];
+    int32_t vnodeIndex = 0;
+    int32_t dnodeIndex = 0;
+    strdequote(pValToken->z);
+    bool parseOk = taosCheckBalanceCfgOptions(pValToken->z, &vnodeIndex, &dnodeIndex);
+    if (!parseOk) {
+      return TSDB_CODE_TSC_INVALID_SQL;  // options value is invalid
+    }
+    return TSDB_CODE_SUCCESS;
+  } else if ((strncasecmp(cfgOptions[tokenMonitor].name, pOptionToken->z, pOptionToken->n) == 0) &&
+             (cfgOptions[tokenMonitor].len == pOptionToken->n)) {
     SSQLToken* pValToken = &pOptions->a[2];
     int32_t    val = strtol(pValToken->z, NULL, 10);
     if (val != 0 && val != 1) {
@@ -4709,8 +4725,8 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
       return TSDB_CODE_TSC_INVALID_SQL;
     }
 
-    for (int32_t i = 2; i < DNODE_DYNAMIC_CFG_OPTIONS_SIZE - 1; ++i) {
-      const SDNodeDynConfOption* pOption = &DNODE_DYNAMIC_CFG_OPTIONS[i];
+    for (int32_t i = tokenDebugFlag; i < tokenDebugFlagEnd; ++i) {
+      const SDNodeDynConfOption* pOption = &cfgOptions[i];
 
       if ((strncasecmp(pOption->name, pOptionToken->z, pOptionToken->n) == 0) && (pOption->len == pOptionToken->n)) {
         /* options is valid */

src/client/src/tscServer.c

@@ -294,52 +294,31 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
     }
   }
 
-  if (rpcMsg->pCont == NULL) {
-    rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
-  } else {
-    STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
-    // if (rpcMsg->code != TSDB_CODE_RPC_NETWORK_UNAVAIL) {
-    //   if (pCmd->command == TSDB_SQL_CONNECT) {
-    //     rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
-    //     rpcFreeCont(rpcMsg->pCont);
-    //     return;
-    //   }
-
-    //   if (pCmd->command == TSDB_SQL_HB) {
-    //     rpcMsg->code = TSDB_CODE_RPC_NOT_READY;
-    //     rpcFreeCont(rpcMsg->pCont);
-    //     return;
-    //   }
-
-    //   if (pCmd->command == TSDB_SQL_META || pCmd->command == TSDB_SQL_DESCRIBE_TABLE ||
-    //       pCmd->command == TSDB_SQL_STABLEVGROUP || pCmd->command == TSDB_SQL_SHOW ||
-    //       pCmd->command == TSDB_SQL_RETRIEVE) {
-    //     // get table meta/vgroup query will not retry, do nothing
-    //   }
-    // }
-
-    if ((pCmd->command == TSDB_SQL_SELECT || pCmd->command == TSDB_SQL_FETCH || pCmd->command == TSDB_SQL_INSERT ||
-         pCmd->command == TSDB_SQL_UPDATE_TAGS_VAL) &&
-        (rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID || rpcMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID ||
-         rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE)) {
-      tscWarn("%p it shall renew table meta, code:%s, retry:%d", pSql, tstrerror(rpcMsg->code), ++pSql->retry);
-      // set the flag to denote that sql string needs to be re-parsed and build submit block with table schema
-      if (rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-        pSql->cmd.submitSchema = 1;
-      }
+  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
 
-      pSql->res.code = rpcMsg->code;  // keep the previous error code
-      if (pSql->retry > pSql->maxRetry) {
-        tscError("%p max retry %d reached, give up", pSql, pSql->maxRetry);
-      } else {
-        rpcMsg->code = tscRenewTableMeta(pSql, pTableMetaInfo->name);
-
-        // if there is an error occurring, proceed to the following error handling procedure.
-        // todo add test cases
-        if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
-          rpcFreeCont(rpcMsg->pCont);
-          return;
-        }
+  int32_t cmd = pCmd->command;
+  if ((cmd == TSDB_SQL_SELECT || cmd == TSDB_SQL_FETCH || cmd == TSDB_SQL_INSERT || cmd == TSDB_SQL_UPDATE_TAGS_VAL) &&
+      (rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID ||
+       rpcMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID ||
+       rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL ||
+       rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE)) {
+    tscWarn("%p it shall renew table meta, code:%s, retry:%d", pSql, tstrerror(rpcMsg->code), ++pSql->retry);
+
+    // set the flag to denote that sql string needs to be re-parsed and build submit block with table schema
+    if (rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
+      pSql->cmd.submitSchema = 1;
+    }
+
+    pSql->res.code = rpcMsg->code;  // keep the previous error code
+    if (pSql->retry > pSql->maxRetry) {
+      tscError("%p max retry %d reached, give up", pSql, pSql->maxRetry);
+    } else {
+      rpcMsg->code = tscRenewTableMeta(pSql, pTableMetaInfo->name);
+
+      // if there is an error occurring, proceed to the following error handling procedure.
+      if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+        rpcFreeCont(rpcMsg->pCont);
+        return;
       }
     }
   }

src/client/src/tscSql.c

@@ -15,7 +15,7 @@
 
 #include "hash.h"
 #include "os.h"
-#include "qast.h"
+#include "qAst.h"
 #include "tcache.h"
 #include "tnote.h"
 #include "trpc.h"
@@ -724,6 +724,13 @@ int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields)
   return len;
 }
 
+static void asyncCallback(void *param, TAOS_RES *tres, int code) {
+  assert(param != NULL);
+  SSqlObj *pSql = ((SSqlObj *)param);
+  pSql->res.code = code;
+  sem_post(&pSql->rspSem);
+}
+
 int taos_validate_sql(TAOS *taos, const char *sql) {
   STscObj *pObj = (STscObj *)taos;
   if (pObj == NULL || pObj->signature != pObj) {
@@ -732,7 +739,8 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
   }
 
   SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
-  
+  pSql->pTscObj = taos;
+  pSql->signature = pSql;
   SSqlRes *pRes = &pSql->res;
   SSqlCmd *pCmd = &pSql->cmd;
   
@@ -766,10 +774,17 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
     pCmd->pTableList = NULL;
   }
 
-  pRes->code = (uint8_t)tsParseSql(pSql, false);
-  int code = pRes->code;
-
-  tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj);
+  pSql->fp = asyncCallback;
+  pSql->fetchFp = asyncCallback;
+  pSql->param = pSql;
+  int code = tsParseSql(pSql, true);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    sem_wait(&pSql->rspSem);
+    code = pSql->res.code;
+  }
+  if (code != TSDB_CODE_SUCCESS) {
+    tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, code, taos_errstr(taos), pObj);
+  }
   taos_free_result(pSql);
 
   return code;
@@ -865,6 +880,8 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
   }
 
   SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
+  pSql->pTscObj = taos;
+  pSql->signature = pSql;
   SSqlRes *pRes = &pSql->res;
 
   pRes->numOfTotal = 0;  // the number of getting table meta from server

src/client/src/tscStream.c

@@ -122,7 +122,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
       pQueryInfo->window.ekey = pStream->etime;
     }
   } else {
-    pQueryInfo->window.skey = pStream->stime - pStream->interval;
+    pQueryInfo->window.skey = pStream->stime;
     int64_t etime = taosGetTimestamp(pStream->precision);
     // delay to wait all data in last time window
     if (pStream->precision == TSDB_TIME_PRECISION_MICRO) {
@@ -232,6 +232,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
       (*pStream->fp)(pStream->param, res, row);
     }
 
+    if (!pStream->isProject) {
+      pStream->stime += pStream->slidingTime;
+    }
     // actually only one row is returned. this following is not necessary
     taos_fetch_rows_a(res, tscProcessStreamRetrieveResult, pStream);
   } else {  // numOfRows == 0, all data has been retrieved
@@ -432,6 +435,7 @@ static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, in
   } else {             // timewindow based aggregation stream
     if (stime == 0) {  // no data in meter till now
       stime = ((int64_t)taosGetTimestamp(pStream->precision) / pStream->interval) * pStream->interval;
+      stime -= pStream->interval;
       tscWarn("%p stream:%p, last timestamp:0, reset to:%" PRId64, pSql, pStream, stime);
     } else {
       int64_t newStime = (stime / pStream->interval) * pStream->interval;

src/client/src/tscSub.c

@@ -34,6 +34,7 @@ typedef struct SSubscriptionProgress {
 typedef struct SSub {
   void *                  signature;
   char                    topic[32];
+  sem_t                   sem;
   int64_t                 lastSyncTime;
   int64_t                 lastConsumeTime;
   TAOS *                  taos;
@@ -83,84 +84,108 @@ void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts) {
 
 static void asyncCallback(void *param, TAOS_RES *tres, int code) {
   assert(param != NULL);
-  SSqlObj *pSql = ((SSqlObj *)param);
-  
-  pSql->res.code = code;
-  sem_post(&pSql->rspSem);
+  SSub *pSub = ((SSub *)param);
+  pSub->pSql->res.code = code;
+  sem_post(&pSub->sem);
 }
 
 
 static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* sql) {
-  SSub* pSub = NULL;
-
-  TRY( 8 ) {
-    SSqlObj* pSql = calloc_throw(1, sizeof(SSqlObj));
-    CLEANUP_PUSH_FREE(true, pSql);
-    SSqlCmd *pCmd = &pSql->cmd;
-    SSqlRes *pRes = &pSql->res;
-
-    if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
-      THROW(TAOS_SYSTEM_ERROR(errno));
-    }
-    CLEANUP_PUSH_INT_PTR(true, tsem_destroy, &pSql->rspSem);
+  int code = TSDB_CODE_SUCCESS, line = __LINE__;
+  SSqlObj* pSql = NULL;
 
-    pSql->signature = pSql;
-    pSql->param = pSql;
-    pSql->pTscObj = pObj;
-    pSql->maxRetry = TSDB_MAX_REPLICA;
-    pSql->fp = asyncCallback;
-
-    int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
-    if (code != TSDB_CODE_SUCCESS) {
-      THROW(code);
-    }
-    CLEANUP_PUSH_FREE(true, pCmd->payload);
+  SSub* pSub = calloc(1, sizeof(SSub));
+  if (pSub == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  pSub->signature = pSub;
+  if (tsem_init(&pSub->sem, 0, 0) == -1) {
+    line = __LINE__;
+    code = TAOS_SYSTEM_ERROR(errno);
+    goto fail;
+  }
+  tstrncpy(pSub->topic, topic, sizeof(pSub->topic));
+  pSub->progress = taosArrayInit(32, sizeof(SSubscriptionProgress));
+  if (pSub->progress == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
 
-    pRes->qhandle = 0;
-    pRes->numOfRows = 1;
+  pSql = calloc(1, sizeof(SSqlObj));
+  if (pSql == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  pSql->signature = pSql;
+  pSql->pTscObj = pObj;
+  pSql->pSubscription = pSub;
+  pSub->pSql = pSql;
 
-    pSql->sqlstr = strdup_throw(sql);
-    CLEANUP_PUSH_FREE(true, pSql->sqlstr);
-    strtolower(pSql->sqlstr, pSql->sqlstr);
+  SSqlCmd* pCmd = &pSql->cmd;
+  SSqlRes* pRes = &pSql->res;
+  if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
+    line = __LINE__;
+    code = TAOS_SYSTEM_ERROR(errno);
+    goto fail;
+  }
 
-    code = tsParseSql(pSql, false);
-    if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
-      // wait for the callback function to post the semaphore
-      sem_wait(&pSql->rspSem);
-      code = pSql->res.code;
-    }
-    if (code != TSDB_CODE_SUCCESS) {
-      tscError("failed to parse sql statement: %s, error: %s", pSub->topic, tstrerror(code));
-      THROW( code );
-    }
+  pSql->param = pSub;
+  pSql->maxRetry = TSDB_MAX_REPLICA;
+  pSql->fp = asyncCallback;
+  pSql->fetchFp = asyncCallback;
+  pSql->sqlstr = strdup(sql);
+  if (pSql->sqlstr == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  strtolower(pSql->sqlstr, pSql->sqlstr);
+  pRes->qhandle = 0;
+  pRes->numOfRows = 1;
+
+  code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
+  if (code != TSDB_CODE_SUCCESS) {
+    line = __LINE__;
+    goto fail;
+  }
 
-    if (pSql->cmd.command != TSDB_SQL_SELECT) {
-      tscError("only 'select' statement is allowed in subscription: %s", pSub->topic);
-      THROW( -1 );  // TODO
-    }
+  code = tsParseSql(pSql, false);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    sem_wait(&pSub->sem);
+    code = pSql->res.code;
+  }
+  if (code != TSDB_CODE_SUCCESS) {
+    line = __LINE__;
+    goto fail;
+  }
 
-    pSub = calloc_throw(1, sizeof(SSub));
-    CLEANUP_PUSH_FREE(true, pSub);
-    pSql->pSubscription = pSub;
-    pSub->pSql = pSql;
-    pSub->signature = pSub;
-    strncpy(pSub->topic, topic, sizeof(pSub->topic));
-    pSub->topic[sizeof(pSub->topic) - 1] = 0;
-    pSub->progress = taosArrayInit(32, sizeof(SSubscriptionProgress));
-    if (pSub->progress == NULL) {
-      THROW(TSDB_CODE_TSC_OUT_OF_MEMORY);
-    }
+  if (pSql->cmd.command != TSDB_SQL_SELECT) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_INVALID_SQL;
+    goto fail;
+  }
 
-    CLEANUP_EXECUTE();
+  return pSub;
 
-  } CATCH( code ) {
-    tscError("failed to create subscription object: %s", tstrerror(code));
-    CLEANUP_EXECUTE();
+fail:
+  tscError("tscCreateSubscription failed at line %d, reason: %s", line, tstrerror(code));
+  if (pSql != NULL) {
+    tscFreeSqlObj(pSql);
+    pSql = NULL;
+  }
+  if (pSub != NULL) {
+    taosArrayDestroy(pSub->progress);
+    tsem_destroy(&pSub->sem);
+    free(pSub);
     pSub = NULL;
+  }
 
-  } END_TRY
-
-  return pSub;
+  terrno = code;
+  return NULL;
 }
 
 
@@ -405,9 +430,10 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
     tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0)->vgroupIndex = 0;
 
     pSql->fp = asyncCallback;
-    pSql->param = pSql;
+    pSql->fetchFp = asyncCallback;
+    pSql->param = pSub;
     tscDoQuery(pSql);
-    sem_wait(&pSql->rspSem);
+    sem_wait(&pSub->sem);
 
     if (pRes->code != TSDB_CODE_SUCCESS) {
       continue;
@@ -437,7 +463,9 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
   }
 
   if (keepProgress) {
-    tscSaveSubscriptionProgress(pSub);
+    if (pSub->progress != NULL) {
+      tscSaveSubscriptionProgress(pSub);
+    }
   } else {
     char path[256];
     sprintf(path, "%s/subscribe/%s", tsDataDir, pSub->topic);
@@ -448,6 +476,7 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
 
   tscFreeSqlObj(pSub->pSql);
   taosArrayDestroy(pSub->progress);
+  tsem_destroy(&pSub->sem);
   memset(pSub, 0, sizeof(*pSub));
   free(pSub);
 }

src/client/src/tscSubquery.c

@@ -14,8 +14,8 @@
  */
 #include "os.h"
 
-#include "qtsbuf.h"
-#include "qast.h"
+#include "qAst.h"
+#include "qTsbuf.h"
 #include "tcompare.h"
 #include "tscLog.h"
 #include "tscSubquery.h"

src/client/src/tscUtil.c

@@ -13,11 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
-#include "hash.h"
 #include "tscUtil.h"
+#include "hash.h"
+#include "os.h"
+#include "qAst.h"
 #include "taosmsg.h"
-#include "qast.h"
 #include "tcache.h"
 #include "tkey.h"
 #include "tmd5.h"

src/common/inc/tglobal.h

@@ -174,6 +174,7 @@ bool taosCheckGlobalCfg();
 void taosSetAllDebugFlag();
 bool taosCfgDynamicOptions(char *msg);
 int  taosGetFqdnPortFromEp(const char *ep, char *fqdn, uint16_t *port);
+bool taosCheckBalanceCfgOptions(const char *option, int32_t *vnodeIndex, int32_t *dnodeIndex);
  
 #ifdef __cplusplus
 }

src/common/src/tdataformat.c

@@ -318,7 +318,7 @@ SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows) {
   pCols->maxPoints = maxRows;
   pCols->bufSize = maxRowSize * maxRows;
 
-  pCols->buf = calloc(1, pCols->bufSize);
+  pCols->buf = malloc(pCols->bufSize);
   if (pCols->buf == NULL) {
     free(pCols);
     return NULL;

src/common/src/tglobal.c

@@ -198,6 +198,7 @@ int32_t tsdbDebugFlag = 131;
 
 int32_t (*monitorStartSystemFp)() = NULL;
 void (*monitorStopSystemFp)() = NULL;
+void (*monitorExecuteSQLFp)(char *sql) = NULL;
 
 static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT;
 
@@ -252,11 +253,15 @@ bool taosCfgDynamicOptions(char *msg) {
         if (monitorStartSystemFp) {
           (*monitorStartSystemFp)();
           uInfo("monitor is enabled");
+        } else {
+          uError("monitor can't be updated, for monitor not initialized");
         }
       } else {
         if (monitorStopSystemFp) {
           (*monitorStopSystemFp)();
           uInfo("monitor is disabled");
+        } else {
+          uError("monitor can't be updated, for monitor not initialized");
         }
       }
       return true;
@@ -276,7 +281,12 @@ bool taosCfgDynamicOptions(char *msg) {
   }
 
   if (strncasecmp(option, "resetQueryCache", 15) == 0) {
-    uError("reset query cache can't be executed, for monitor not initialized");
+    if (monitorExecuteSQLFp) {
+      (*monitorExecuteSQLFp)("resetQueryCache");
+      uInfo("resetquerycache is executed");
+    } else {
+      uError("resetquerycache can't be executed, for monitor not started");
+    }
   }
 
   return false;
@@ -1300,3 +1310,32 @@ int taosGetFqdnPortFromEp(const char *ep, char *fqdn, uint16_t *port) {
 
   return 0; 
 }
+
+/*
+ * alter dnode 1 balance "vnode:1-dnode:2"
+ */
+
+bool taosCheckBalanceCfgOptions(const char *option, int32_t *vnodeIndex, int32_t *dnodeIndex) {
+  int len = strlen(option);
+  if (strncasecmp(option, "vnode:", 6) != 0) {
+    return false;
+  }
+
+  int pos = 0;
+  for (; pos < len; ++pos) {
+    if (option[pos] == '-') break;
+  }
+
+  if (++pos >= len) return false;
+  if (strncasecmp(option + pos, "dnode:", 6) != 0) {
+    return false;
+  }
+
+  *vnodeIndex = strtol(option + 6, NULL, 10);
+  *dnodeIndex = strtol(option + pos + 6, NULL, 10);
+  if (*vnodeIndex <= 1 || *dnodeIndex <= 0) {
+    return false;
+  }
+
+  return true;
+}
\ No newline at end of file

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TDNode.java

 package com.taosdata.jdbc.utils;
 
+import java.io.BufferedReader;
 import java.io.File;
+import java.io.InputStreamReader;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 
@@ -31,6 +33,10 @@ public class TDNode {
         this.testCluster = testCluster;
     }
 
+    public void setRunning(int running) {
+        this.running = running;
+    }
+
     public void searchTaosd(File dir, ArrayList<String> taosdPath) {        
         File[] fileList = dir.listFiles();
         
@@ -102,15 +108,46 @@ public class TDNode {
         this.running = 1;        
     }
 
-    public void stop() {
-        String toBeKilled = "taosd";
+    public Integer getTaosdPid() {
+        String cmd = "ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'"; 
+        String[] cmds = {"sh", "-c", cmd};
+        try {
+            Process process = Runtime.getRuntime().exec(cmds);
+            BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+            String line = null;
+            Integer res = null;
+            while((line = reader.readLine()) != null) {
+                if(!line.isEmpty()) {
+                    res = Integer.valueOf(line);
+                    break;
+                }
+            }
+            
+            return res;
+        } catch (Exception e) {
+            e.printStackTrace();
+        } 
+        return null;
+    }
+
+    public void stop() {        
 
         if (this.running != 0) {
-            String killCmd = "pkill -kill -x " + toBeKilled;
-            String[] killCmds = {"sh", "-c", killCmd};
-            try {                                
-                Runtime.getRuntime().exec(killCmds).waitFor();
+            Integer pid = null;                        
+            while((pid = getTaosdPid()) != null) {
+                
+                String killCmd = "kill -term " + pid;                                      
+                String[] killCmds = {"sh", "-c", killCmd};
+                try {                                
+                    Runtime.getRuntime().exec(killCmds).waitFor();                    
+
+                    TimeUnit.SECONDS.sleep(2);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }                
+            }
 
+            try {
                 for(int port = 6030; port < 6041; port ++) {
                     String fuserCmd = "fuser -k -n tcp " + port;
                     Runtime.getRuntime().exec(fuserCmd).waitFor();
@@ -120,7 +157,7 @@ public class TDNode {
             }
 
             this.running = 0;
-            System.out.println("dnode:" + this.index + " is stopped by pkill");
+            System.out.println("dnode:" + this.index + " is stopped by kill -term");
         }
     }
 

src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TDNodes.java

@@ -14,33 +14,6 @@ public class TDNodes {
         }               
     }
 
-    public void setPath(String path) {
-        try {
-            String killCmd = "pkill -kill -x taosd";
-            String[] killCmds = {"sh", "-c", killCmd};
-            Runtime.getRuntime().exec(killCmds).waitFor();        
-
-            String binPath = System.getProperty("user.dir");
-            binPath += "/../../../debug";
-            System.out.println("binPath: " + binPath);
-        
-            File file = new File(path);
-            binPath = file.getCanonicalPath();
-            System.out.println("binPath real path: " + binPath);
-                
-            if(path.isEmpty()){
-                file = new File(path + "/../../");                
-                path = file.getCanonicalPath();
-            }            
-
-            for(int i = 0; i < tdNodes.size(); i++) {
-                tdNodes.get(i).setPath(path);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-    
     public void setTestCluster(boolean testCluster) {
         this.testCluster = testCluster;
     }
@@ -70,6 +43,11 @@ public class TDNodes {
         check(index);
         tdNodes.get(index - 1).setCfgConfig(option, value);
     }
+
+    public TDNode getTDNode(int index) {
+        check(index);
+        return tdNodes.get(index - 1);
+    }
     
     public void start(int index) {
         check(index);

src/connector/jdbc/src/test/java/com/taosdata/jdbc/BaseTest.java

 package com.taosdata.jdbc;
 
-import java.io.File;
 import com.taosdata.jdbc.utils.TDNodes;
 
 import org.junit.AfterClass;
@@ -8,31 +7,30 @@ import org.junit.BeforeClass;
 
 public class BaseTest {
 
-    private static boolean testCluster = false;
-    private static String deployPath = System.getProperty("user.dir");         
-    private static TDNodes tdNodes = new TDNodes();    
+    private static boolean testCluster = false;           
+    private static TDNodes nodes = new TDNodes();    
     
     
     @BeforeClass
     public static void setupEnv() {
-        try{
-            File file = new File(deployPath + "/../../../");
-            String rootPath = file.getCanonicalPath();
-                       
-            tdNodes.setPath(rootPath);
-            tdNodes.setTestCluster(testCluster);            
+        try{                         
+            if (nodes.getTDNode(1).getTaosdPid() != null) {
+                System.out.println("Kill taosd before running JDBC test");
+                nodes.getTDNode(1).setRunning(1);
+                nodes.stop(1);
+            }
 
-            tdNodes.deploy(1);
-            tdNodes.start(1);  
+            nodes.setTestCluster(testCluster);  
+            nodes.deploy(1);
+            nodes.start(1);  
 
         } catch (Exception e) {
-            e.printStackTrace();
-            System.out.println("Base Test Exception");
+            e.printStackTrace();            
         }
     }
 
     @AfterClass
     public static void cleanUpEnv() {
-        tdNodes.stop(1);
+        nodes.stop(1);
     }
 }
\ No newline at end of file

src/inc/taos.h

@@ -67,8 +67,6 @@ DLL_EXPORT void  taos_init();
 DLL_EXPORT void  taos_cleanup();
 DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
 DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_c(const char *ip, uint8_t ipLen, const char *user, uint8_t userLen,
-                 const char *pass, uint8_t passLen, const char *db, uint8_t dbLen, uint16_t port);
 DLL_EXPORT void  taos_close(TAOS *taos);
 
 typedef struct TAOS_BIND {
@@ -90,7 +88,6 @@ TAOS_RES * taos_stmt_use_result(TAOS_STMT *stmt);
 int        taos_stmt_close(TAOS_STMT *stmt);
 
 DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
-DLL_EXPORT TAOS_RES *taos_query_c(TAOS *taos, const char *sql, uint32_t sqlLen);
 DLL_EXPORT TAOS_ROW taos_fetch_row(TAOS_RES *res);
 DLL_EXPORT int taos_result_precision(TAOS_RES *res);  // get the time precision of result
 DLL_EXPORT void taos_free_result(TAOS_RES *res);

src/inc/taoserror.h

@@ -227,6 +227,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_CPU_LIMITED,            0, 0x080B, "grant cpu
 
 // sync
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CONFIG,           0, 0x0900, "sync invalid configuration")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_ENABLED,              0, 0x0901, "sync module not enabled")
 
 // wal
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR,                0, 0x1000, "wal app error")

src/inc/tbalance.h

@@ -29,6 +29,7 @@ void    balanceAsyncNotify();
 void    balanceSyncNotify();
 void    balanceReset();
 int32_t balanceAllocVnodes(struct SVgObj *pVgroup);
+int32_t balanceCfgDnode(struct SDnodeObj *pDnode, const char *option);
 int32_t balanceDropDnode(struct SDnodeObj *pDnode);
 
 #ifdef __cplusplus

src/kit/taosdemo/taosdemo.c

@@ -361,7 +361,7 @@ int main(int argc, char *argv[]) {
   arguments.num_of_DPT = 100000;
   arguments.num_of_RPR = 1000;
   arguments.use_metric = true;
-  arguments.insert_only = true;
+  arguments.insert_only = false;
   // end change
 
   argp_parse(&argp, argc, argv, 0, 0, &arguments);
@@ -954,13 +954,13 @@ void *readMetric(void *sarg) {
 
     for (int i = 1; i <= m; i++) {
       if (i == 1) {
-        sprintf(tempS, "index = %d", i);
+        sprintf(tempS, "areaid = %d", i);
       } else {
-        sprintf(tempS, " or index = %d ", i);
+        sprintf(tempS, " or areaid = %d ", i);
       }
       strcat(condition, tempS);
 
-      sprintf(command, "select %s from m1 where %s", aggreFunc[j], condition);
+      sprintf(command, "select %s from meters where %s", aggreFunc[j], condition);
 
       printf("Where condition: %s\n", condition);
       fprintf(fp, "%s\n", command);

src/mnode/inc/mnodeDef.h

@@ -50,8 +50,8 @@ typedef struct SDnodeObj {
   int8_t     alternativeRole;  // from dnode status msg, 0-any, 1-mgmt, 2-dnode
   int8_t     status;           // set in balance function
   int8_t     isMgmt;
-  int8_t     reserve1[14];  
-  int8_t     updateEnd[1];
+  int8_t     reserve1[11];  
+  int8_t     updateEnd[4];
   int32_t    refCount;
   uint32_t   moduleStatus;
   uint32_t   lastReboot;       // time stamp for last reboot
@@ -68,8 +68,8 @@ typedef struct SMnodeObj {
   int32_t    mnodeId;
   int8_t     reserved0[4];
   int64_t    createdTime;
-  int8_t     reserved1[7];
-  int8_t     updateEnd[1];
+  int8_t     reserved1[4];
+  int8_t     updateEnd[4];
   int32_t    refCount;
   int8_t     role;
   int8_t     reserved2[3];
@@ -90,8 +90,7 @@ typedef struct SSuperTableObj {
   int32_t    tversion;
   int32_t    numOfColumns;
   int32_t    numOfTags;
-  int8_t     reserved1[3];
-  int8_t     updateEnd[1];
+  int8_t     updateEnd[4];
   int32_t    refCount;
   int32_t    numOfTables;
   SSchema *  schema;
@@ -111,8 +110,7 @@ typedef struct {
   int32_t    sid;
   int32_t    vgId;
   int32_t    sqlLen;
-  int8_t     updateEnd[1];
-  int8_t     reserved1[1]; 
+  int8_t     updateEnd[4];
   int32_t    refCount;
   char*      sql;          //used by normal table
   SSchema*   schema;       //used by normal table
@@ -138,8 +136,8 @@ typedef struct SVgObj {
   int8_t         status;
   int8_t         reserved0[4];
   SVnodeGid      vnodeGid[TSDB_MAX_REPLICA];
-  int8_t         reserved1[7];
-  int8_t         updateEnd[1];
+  int8_t         reserved1[4];
+  int8_t         updateEnd[4];
   int32_t        refCount;
   int32_t        numOfTables;
   int64_t        totalStorage;
@@ -176,8 +174,8 @@ typedef struct SDbObj {
   int32_t cfgVersion;
   SDbCfg  cfg;
   int8_t  status;
-  int8_t  reserved1[14];
-  int8_t  updateEnd[1];
+  int8_t  reserved1[11];
+  int8_t  updateEnd[4];
   int32_t refCount;
   int32_t numOfVgroups;
   int32_t numOfTables;
@@ -196,8 +194,8 @@ typedef struct SUserObj {
   int64_t           createdTime;
   int8_t            superAuth;
   int8_t            writeAuth;
-  int8_t            reserved[13];
-  int8_t            updateEnd[1];
+  int8_t            reserved[10];
+  int8_t            updateEnd[4];
   int32_t           refCount;
   struct SAcctObj * pAcct;
 } SUserObj;
@@ -228,11 +226,11 @@ typedef struct SAcctObj {
   int64_t   createdTime;
   int32_t   acctId;
   int8_t    status;
-  int8_t    reserved0[10];
-  int8_t    updateEnd[1];
-  SAcctInfo acctInfo;
+  int8_t    reserved0[7];
+  int8_t    updateEnd[4];
   int32_t   refCount;
   int8_t    reserved1[4];
+  SAcctInfo acctInfo;
   pthread_mutex_t  mutex;
 } SAcctObj;
 

src/mnode/src/mnodeBalance.c

@@ -28,6 +28,7 @@ void    balanceCleanUp() {}
 void    balanceAsyncNotify() {}
 void    balanceSyncNotify() {}
 void    balanceReset() {}
+int32_t balanceCfgDnode(struct SDnodeObj *pDnode, const char *option) { return TSDB_CODE_SYN_NOT_ENABLED; }
 
 int32_t balanceAllocVnodes(SVgObj *pVgroup) {
   void *     pIter = NULL;

src/mnode/src/mnodeDb.c

@@ -67,8 +67,11 @@ static int32_t mnodeDbActionInsert(SSdbOper *pOper) {
   SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
 
   pthread_mutex_init(&pDb->mutex, NULL);
+  pthread_mutex_lock(&pDb->mutex);
   pDb->vgListSize = VG_LIST_SIZE;
   pDb->vgList = calloc(pDb->vgListSize, sizeof(SVgObj *));
+  pthread_mutex_unlock(&pDb->mutex);
+
   pDb->numOfVgroups = 0;
   pDb->numOfTables = 0;
   pDb->numOfSuperTables = 0;
@@ -395,8 +398,8 @@ static int32_t mnodeCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate, void *pMs
 
   code = sdbInsertRow(&oper);
   if (code != TSDB_CODE_SUCCESS) {
-    mnodeDestroyDb(pDb);
     mLInfo("db:%s, failed to create, reason:%s", pDb->name, tstrerror(code));
+    mnodeDestroyDb(pDb);
     return code;
   } else {
     return TSDB_CODE_MND_ACTION_IN_PROGRESS;
@@ -605,7 +608,9 @@ static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn
 
 static char *mnodeGetDbStr(char *src) {
   char *pos = strstr(src, TS_PATH_DELIMITER);
-  return ++pos;
+  if (pos != NULL) ++pos;
+
+  return pos;
 }
 
 static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
@@ -622,10 +627,13 @@ static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void
 
     cols = 0;
 
-    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;  
     char* name = mnodeGetDbStr(pDb->name);
-    STR_WITH_MAXSIZE_TO_VARSTR(pWrite, name, pShow->bytes[cols]);
+    if (name != NULL) {
+      STR_WITH_MAXSIZE_TO_VARSTR(pWrite, name, pShow->bytes[cols]);
+    } else {
+      STR_TO_VARSTR(pWrite, "NULL");
+    }
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;

src/mnode/src/mnodeDnode.c

@@ -277,45 +277,45 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
   SCMCfgDnodeMsg *pCmCfgDnode = pMsg->rpcMsg.pCont;
   if (pCmCfgDnode->ep[0] == 0) {
     tstrncpy(pCmCfgDnode->ep, tsLocalEp, TSDB_EP_LEN);
-  } 
+  }
 
-  int32_t dnodeId = 0;
-  char* pos = strchr(pCmCfgDnode->ep, ':');
-  if (NULL == pos) {
-    dnodeId = strtol(pCmCfgDnode->ep, NULL, 10);
+  SDnodeObj *pDnode = mnodeGetDnodeByEp(pCmCfgDnode->ep);
+  if (pDnode == NULL) {
+    int32_t dnodeId = strtol(pCmCfgDnode->ep, NULL, 10);
     if (dnodeId <= 0 || dnodeId > 65536) {
-      mError("failed to cfg dnode, invalid dnodeId:%s", pCmCfgDnode->ep);
+      mError("failed to cfg dnode, invalid dnodeEp:%s", pCmCfgDnode->ep);
       return TSDB_CODE_MND_DNODE_NOT_EXIST;
     }
-  }
 
-  SRpcEpSet epSet = mnodeGetEpSetFromIp(pCmCfgDnode->ep);
-  if (dnodeId != 0) {
-    SDnodeObj *pDnode = mnodeGetDnode(dnodeId);
+    pDnode = mnodeGetDnode(dnodeId);
     if (pDnode == NULL) {
       mError("failed to cfg dnode, invalid dnodeId:%d", dnodeId);
       return TSDB_CODE_MND_DNODE_NOT_EXIST;
     }
-    epSet = mnodeGetEpSetFromIp(pDnode->dnodeEp);
-    mnodeDecDnodeRef(pDnode);
   }
 
-  SMDCfgDnodeMsg *pMdCfgDnode = rpcMallocCont(sizeof(SMDCfgDnodeMsg));
-  strcpy(pMdCfgDnode->ep, pCmCfgDnode->ep);
-  strcpy(pMdCfgDnode->config, pCmCfgDnode->config);
-
-  SRpcMsg rpcMdCfgDnodeMsg = {
-    .ahandle = 0,
-    .code = 0,
-    .msgType = TSDB_MSG_TYPE_MD_CONFIG_DNODE,
-    .pCont = pMdCfgDnode,
-    .contLen = sizeof(SMDCfgDnodeMsg)
-  };
-
-  mInfo("dnode:%s, is configured by %s", pCmCfgDnode->ep, pMsg->pUser->user);
-  dnodeSendMsgToDnode(&epSet, &rpcMdCfgDnodeMsg);
+  SRpcEpSet epSet = mnodeGetEpSetFromIp(pDnode->dnodeEp);
+  mnodeDecDnodeRef(pDnode);
 
-  return TSDB_CODE_SUCCESS;
+  if (strncasecmp(pCmCfgDnode->config, "balance", 7) == 0) {
+    return balanceCfgDnode(pDnode, pCmCfgDnode->config + 8);
+  } else {
+    SMDCfgDnodeMsg *pMdCfgDnode = rpcMallocCont(sizeof(SMDCfgDnodeMsg));
+    strcpy(pMdCfgDnode->ep, pCmCfgDnode->ep);
+    strcpy(pMdCfgDnode->config, pCmCfgDnode->config);
+
+    SRpcMsg rpcMdCfgDnodeMsg = {
+      .ahandle = 0,
+      .code = 0,
+      .msgType = TSDB_MSG_TYPE_MD_CONFIG_DNODE,
+      .pCont = pMdCfgDnode,
+      .contLen = sizeof(SMDCfgDnodeMsg)
+    };
+
+    mInfo("dnode:%s, is configured by %s", pCmCfgDnode->ep, pMsg->pUser->user);
+    dnodeSendMsgToDnode(&epSet, &rpcMdCfgDnodeMsg);
+    return TSDB_CODE_SUCCESS;
+  }
 }
 
 static void mnodeProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) {

src/mnode/src/mnodeSdb.c

@@ -406,7 +406,7 @@ void sdbDecRef(void *handle, void *pObj) {
   int32_t    refCount = atomic_sub_fetch_32(pRefCount, 1);
   sdbTrace("def ref of table:%s record:%p:%s:%d", pTable->tableName, pObj, sdbGetKeyStrFromObj(pTable, pObj), *pRefCount);
 
-  int8_t *updateEnd = pObj + pTable->refCountPos - 1;
+  int32_t *updateEnd = pObj + pTable->refCountPos - 4;
   if (refCount <= 0 && *updateEnd) {
     sdbTrace("table:%s, record:%p:%s:%d is destroyed", pTable->tableName, pObj, sdbGetKeyStrFromObj(pTable, pObj), *pRefCount);
     SSdbOper oper = {.pObj = pObj};
@@ -453,7 +453,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
     keySize = strlen((char *)key);
   }
 
-  taosHashPut(pTable->iHandle, key, keySize, &pOper->pObj, sizeof(void **));
+  taosHashPut(pTable->iHandle, key, keySize, &pOper->pObj, sizeof(int64_t));
 
   sdbIncRef(pTable, pOper->pObj);
   atomic_add_fetch_32(&pTable->numOfRows, 1);
@@ -472,6 +472,14 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
 }
 
 static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
+  int32_t *updateEnd = pOper->pObj + pTable->refCountPos - 4;
+  bool set = atomic_val_compare_exchange_32(updateEnd, 0, 1) == 0;
+  if (!set) {
+    sdbError("table:%s, failed to delete record:%s from hash, for it already removed", pTable->tableName,
+             sdbGetKeyStrFromObj(pTable, pOper->pObj));
+    return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
+  }
+
   (*pTable->deleteFp)(pOper);
   
   void *  key = sdbGetObjKey(pTable, pOper->pObj);
@@ -486,8 +494,6 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
   sdbDebug("table:%s, delete record:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
            sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, pOper->pMsg);
 
-  int8_t *updateEnd = pOper->pObj + pTable->refCountPos - 1;
-  *updateEnd = 1;
   sdbDecRef(pTable, pOper->pObj);
 
   return TSDB_CODE_SUCCESS;
@@ -654,8 +660,9 @@ bool sdbCheckRowDeleted(void *pTableInput, void *pRow) {
   SSdbTable *pTable = pTableInput;
   if (pTable == NULL) return false;
 
-  int8_t *updateEnd = pRow + pTable->refCountPos - 1;
-  return (*updateEnd == 1);
+  int32_t *updateEnd = pRow + pTable->refCountPos - 4;
+  return atomic_val_compare_exchange_32(updateEnd, 1, 1) == 1;
+  // return (*updateEnd == 1);
 }
 
 int32_t sdbDeleteRow(SSdbOper *pOper) {

src/mnode/src/mnodeShow.c

@@ -236,7 +236,7 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
   }
 
   SCMHeartBeatMsg *pHBMsg = pMsg->rpcMsg.pCont;
-  SRpcConnInfo connInfo;
+  SRpcConnInfo connInfo = {0};
   rpcGetConnInfo(pMsg->rpcMsg.handle, &connInfo);
     
   int32_t connId = htonl(pHBMsg->connId);
@@ -284,7 +284,7 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
   SCMConnectRsp *pConnectRsp = NULL;
   int32_t code = TSDB_CODE_SUCCESS;
 
-  SRpcConnInfo connInfo;
+  SRpcConnInfo connInfo = {0};
   if (rpcGetConnInfo(pMsg->rpcMsg.handle, &connInfo) != 0) {
     mError("thandle:%p is already released while process connect msg", pMsg->rpcMsg.handle);
     code = TSDB_CODE_MND_INVALID_CONNECTION;

src/mnode/src/mnodeTable.c

@@ -72,7 +72,7 @@ static void    mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg);
 static int32_t mnodeProcessDropTableMsg(SMnodeMsg *mnodeMsg);
 static int32_t mnodeProcessDropSuperTableMsg(SMnodeMsg *pMsg);
 static void    mnodeProcessDropSuperTableRsp(SRpcMsg *rpcMsg);
-static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn);
+static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg);
 static void    mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg);
 
 static int32_t mnodeProcessSuperTableVgroupMsg(SMnodeMsg *mnodeMsg);
@@ -759,7 +759,7 @@ static int32_t mnodeProcessDropTableMsg(SMnodeMsg *pMsg) {
     SChildTableObj *pCTable = (SChildTableObj *)pMsg->pTable;
     mInfo("app:%p:%p, table:%s, start to drop ctable, vgId:%d sid:%d uid:%" PRIu64, pMsg->rpcMsg.ahandle, pMsg,
           pDrop->tableId, pCTable->vgId, pCTable->sid, pCTable->uid);
-    return mnodeProcessDropChildTableMsg(pMsg, true);
+    return mnodeProcessDropChildTableMsg(pMsg);
   }
 }
 
@@ -882,7 +882,7 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) {
 static int32_t mnodeDropSuperTableCb(SMnodeMsg *pMsg, int32_t code) {
   SSuperTableObj *pTable = (SSuperTableObj *)pMsg->pTable;
   if (code != TSDB_CODE_SUCCESS) {
-    mError("app:%p:%p, table:%s, failed to drop, sdb error", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId);
+    mError("app:%p:%p, stable:%s, failed to drop, sdb error", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId);
   } else {
     mLInfo("app:%p:%p, stable:%s, is dropped from sdb", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId);
   }
@@ -1765,18 +1765,13 @@ static int32_t mnodeProcessCreateChildTableMsg(SMnodeMsg *pMsg) {
   }
 }
 
-static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
+static int32_t mnodeSendDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
   SChildTableObj *pTable = (SChildTableObj *)pMsg->pTable;
-  if (pMsg->pVgroup == NULL) pMsg->pVgroup = mnodeGetVgroup(pTable->vgId);
-  if (pMsg->pVgroup == NULL) {
-    mError("app:%p:%p, table:%s, failed to drop ctable, vgroup not exist", pMsg->rpcMsg.ahandle, pMsg,
-           pTable->info.tableId);
-    return TSDB_CODE_MND_APP_ERROR;
-  }
+  mLInfo("app:%p:%p, ctable:%s, is dropped from sdb", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId);
 
   SMDDropTableMsg *pDrop = rpcMallocCont(sizeof(SMDDropTableMsg));
   if (pDrop == NULL) {
-    mError("app:%p:%p, table:%s, failed to drop ctable, no enough memory", pMsg->rpcMsg.ahandle, pMsg,
+    mError("app:%p:%p, ctable:%s, failed to drop ctable, no enough memory", pMsg->rpcMsg.ahandle, pMsg,
            pTable->info.tableId);
     return TSDB_CODE_MND_OUT_OF_MEMORY;
   }
@@ -1789,7 +1784,7 @@ static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
 
   SRpcEpSet epSet = mnodeGetEpSetFromVgroup(pMsg->pVgroup);
 
-  mInfo("app:%p:%p, table:%s, send drop ctable msg, vgId:%d sid:%d uid:%" PRIu64, pMsg->rpcMsg.ahandle, pMsg,
+  mInfo("app:%p:%p, ctable:%s, send drop ctable msg, vgId:%d sid:%d uid:%" PRIu64, pMsg->rpcMsg.ahandle, pMsg,
         pDrop->tableId, pTable->vgId, pTable->sid, pTable->uid);
 
   SRpcMsg rpcMsg = {
@@ -1807,6 +1802,40 @@ static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg, bool needReturn) {
   return TSDB_CODE_MND_ACTION_IN_PROGRESS;
 }
 
+static int32_t mnodeDropChildTableCb(SMnodeMsg *pMsg, int32_t code) {
+  if (code != TSDB_CODE_SUCCESS) {
+    SChildTableObj *pTable = (SChildTableObj *)pMsg->pTable;
+    mError("app:%p:%p, ctable:%s, failed to drop, sdb error", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId);
+    return code;
+  } 
+
+  return mnodeSendDropChildTableMsg(pMsg, true);
+}
+
+static int32_t mnodeProcessDropChildTableMsg(SMnodeMsg *pMsg) {
+  SChildTableObj *pTable = (SChildTableObj *)pMsg->pTable;
+  if (pMsg->pVgroup == NULL) pMsg->pVgroup = mnodeGetVgroup(pTable->vgId);
+  if (pMsg->pVgroup == NULL) {
+    mError("app:%p:%p, table:%s, failed to drop ctable, vgroup not exist", pMsg->rpcMsg.ahandle, pMsg,
+           pTable->info.tableId);
+    return TSDB_CODE_MND_APP_ERROR;
+  }
+
+  SSdbOper oper = {
+    .type  = SDB_OPER_GLOBAL,
+    .table = tsChildTableSdb,
+    .pObj  = pTable,
+    .pMsg  = pMsg,
+    .cb    = mnodeDropChildTableCb
+  };
+
+  int32_t code = sdbDeleteRow(&oper);
+  if (code == TSDB_CODE_SUCCESS) {
+    return TSDB_CODE_MND_ACTION_IN_PROGRESS;
+  }
+  return code;
+}
+
 static int32_t mnodeFindNormalTableColumnIndex(SChildTableObj *pTable, char *colName) {
   SSchema *schema = (SSchema *) pTable->schema;
   for (int32_t col = 0; col < pTable->numOfColumns; col++) {
@@ -2220,19 +2249,6 @@ static void mnodeProcessDropChildTableRsp(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SSdbOper oper = {
-    .type = SDB_OPER_GLOBAL,
-    .table = tsChildTableSdb,
-    .pObj = pTable
-  };
-
-  int32_t code = sdbDeleteRow(&oper);
-  if (code != TSDB_CODE_SUCCESS) {
-    mError("app:%p:%p, table:%s, update ctables sdb error", mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId);
-    dnodeSendRpcMnodeWriteRsp(mnodeMsg, TSDB_CODE_MND_SDB_ERROR);
-    return;
-  }
-
   if (mnodeMsg->pVgroup->numOfTables <= 0) {
     mInfo("app:%p:%p, vgId:%d, all tables is dropped, drop vgroup", mnodeMsg->rpcMsg.ahandle, mnodeMsg,
            mnodeMsg->pVgroup->vgId);
@@ -2259,7 +2275,7 @@ static void mnodeProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
   if (sdbCheckRowDeleted(tsChildTableSdb, pTable)) {
     mDebug("app:%p:%p, table:%s, create table rsp received, but a deleting opertion incoming, vgId:%d sid:%d uid:%" PRIu64,
            mnodeMsg->rpcMsg.ahandle, mnodeMsg, pTable->info.tableId, pTable->vgId, pTable->sid, pTable->uid);
-    mnodeProcessDropChildTableMsg(mnodeMsg, false);
+    mnodeSendDropChildTableMsg(mnodeMsg, false);
     rpcMsg->code = TSDB_CODE_SUCCESS;
   }
 

src/mnode/src/mnodeUser.c

@@ -358,7 +358,7 @@ static int32_t mnodeRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, voi
 }
 
 SUserObj *mnodeGetUserFromConn(void *pConn) {
-  SRpcConnInfo connInfo;
+  SRpcConnInfo connInfo = {0};
   if (rpcGetConnInfo(pConn, &connInfo) == 0) {
     return mnodeGetUser(connInfo.user);
   } else {

src/mnode/src/mnodeVgroup.c

@@ -434,15 +434,22 @@ int32_t mnodeGetAvailableVgroup(SMnodeMsg *pMsg, SVgObj **ppVgroup, int32_t *pSi
   }
 
   if (pDb->numOfVgroups < maxVgroupsPerDb) {
-    mDebug("app:%p:%p, db:%s, try to create a new vgroup, numOfVgroups:%d maxVgroupsPerDb:%d", pMsg->rpcMsg.ahandle, pMsg,
-           pDb->name, pDb->numOfVgroups, maxVgroupsPerDb);
+    mDebug("app:%p:%p, db:%s, try to create a new vgroup, numOfVgroups:%d maxVgroupsPerDb:%d", pMsg->rpcMsg.ahandle,
+           pMsg, pDb->name, pDb->numOfVgroups, maxVgroupsPerDb);
     pthread_mutex_unlock(&pDb->mutex);
     int32_t code = mnodeCreateVgroup(pMsg);
-    if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return code;
+    if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) {
+      return code;
+    } else {
+      pthread_mutex_lock(&pDb->mutex);
+    }
   }
 
   SVgObj *pVgroup = pDb->vgList[0];
-  if (pVgroup == NULL) return TSDB_CODE_MND_NO_ENOUGH_DNODES;
+  if (pVgroup == NULL) {
+    pthread_mutex_unlock(&pDb->mutex);
+    return TSDB_CODE_MND_NO_ENOUGH_DNODES;
+  }
 
   int32_t code = mnodeAllocVgroupIdPool(pVgroup);
   if (code != TSDB_CODE_SUCCESS) {
@@ -483,7 +490,7 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
   } else {
     pVgroup->status = TAOS_VG_STATUS_READY;
     SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb};
-    sdbUpdateRow(&desc);
+    (void)sdbUpdateRow(&desc);
   }
 
   mInfo("app:%p:%p, vgId:%d, is created in mnode, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,

src/plugins/http/src/gcJson.c

@@ -121,6 +121,10 @@ bool gcBuildQueryJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
 
   for (int k = 0; k < numOfRows; ++k) {
     TAOS_ROW row = taos_fetch_row(result);
+    if (row == NULL) {
+      cmd->numOfRows--;
+      continue;
+    }
     int32_t* length = taos_fetch_lengths(result);
 
     // for group by

src/plugins/http/src/httpContext.c

@@ -108,7 +108,7 @@ HttpContext *httpCreateContext(int32_t fd) {
   pContext->lastAccessTime = taosGetTimestampSec();
   pContext->state = HTTP_CONTEXT_STATE_READY;
 
-  HttpContext **ppContext = taosCachePut(tsHttpServer.contextCache, &pContext, sizeof(void *), &pContext, sizeof(void *), 3);
+  HttpContext **ppContext = taosCachePut(tsHttpServer.contextCache, &pContext, sizeof(int64_t), &pContext, sizeof(int64_t), 3);
   pContext->ppContext = ppContext;
   httpDebug("context:%p, fd:%d, is created, data:%p", pContext, fd, ppContext);
 

src/plugins/http/src/restJson.c

@@ -94,6 +94,10 @@ bool restBuildSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result,
 
   for (int k = 0; k < numOfRows; ++k) {
     TAOS_ROW row = taos_fetch_row(result);
+    if (row == NULL) {
+      cmd->numOfRows--;
+      continue;
+    }
     int32_t* length = taos_fetch_lengths(result);
 
     // data row array begin

src/plugins/monitor/src/monitorMain.c

@@ -27,7 +27,6 @@
 #include "dnode.h"
 #include "monitor.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__); }}
 #define monitorWarn(...)  { if (monitorDebugFlag & DEBUG_WARN)  { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }}
@@ -78,6 +77,7 @@ static void monitorStartTimer();
 static void monitorSaveSystemInfo();
 extern int32_t (*monitorStartSystemFp)();
 extern void (*monitorStopSystemFp)();
+extern void (*monitorExecuteSQLFp)(char *sql);
 
 static void monitorCheckDiskUsage(void *para, void *unused) {
   taosGetDisk();
@@ -207,6 +207,7 @@ static void monitorInitDatabase() {
     taos_query_a(tsMonitorConn.conn, tsMonitorConn.sql, monitorInitDatabaseCb, NULL);
   } else {
     tsMonitorConn.state = MONITOR_STATE_INITIALIZED;
+    monitorExecuteSQLFp = monitorExecuteSQL;
     monitorInfo("monitor service init success");
 
     monitorStartTimer();
@@ -230,6 +231,7 @@ static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code) {
 
 void monitorStopSystem() {
   monitorInfo("monitor module is stopped");
+  monitorExecuteSQLFp = NULL;
   tsMonitorConn.state = MONITOR_STATE_STOPPED;
   if (tsMonitorConn.initTimer != NULL) {
     taosTmrStopA(&(tsMonitorConn.initTimer));
@@ -248,33 +250,13 @@ static void monitorStartTimer() {
   taosTmrReset(monitorSaveSystemInfo, tsMonitorInterval * 1000, NULL, tscTmr, &tsMonitorConn.timer);
 }
 
-static void dnodeMontiorInsertAcctCallback(void *param, TAOS_RES *result, int32_t code) {
-  if (code < 0) {
-    monitorError("monitor:%p, save account info failed, code:%s", tsMonitorConn.conn, tstrerror(code));
-  } else if (code == 0) {
-    monitorError("monitor:%p, save account info failed, affect rows:%d", tsMonitorConn.conn, code);
-  } else {
-    monitorDebug("monitor:%p, save account info success, code:%s", tsMonitorConn.conn, tstrerror(code));
-  }
-}
-
-static void dnodeMontiorInsertSysCallback(void *param, TAOS_RES *result, int32_t code) {
+static void dnodeMontiorLogCallback(void *param, TAOS_RES *result, int32_t code) {
   if (code < 0) {
-    monitorError("monitor:%p, save system info failed, code:%s %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
+    monitorError("monitor:%p, save %s failed, reason:%s", tsMonitorConn.conn, (char *)param, tstrerror(code));
   } else if (code == 0) {
-    monitorError("monitor:%p, save system info failed, affect rows:%d %s", tsMonitorConn.conn, code, tsMonitorConn.sql);
+    monitorError("monitor:%p, save %s failed, affect rows:%d", tsMonitorConn.conn, (char *)param, code);
   } else {
-    monitorDebug("monitor:%p, save system info success, code:%s %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
-  }
-}
-
-static void dnodeMontiorInsertLogCallback(void *param, TAOS_RES *result, int32_t code) {
-  if (code < 0) {
-    monitorError("monitor:%p, save log failed, code:%s", tsMonitorConn.conn, tstrerror(code));
-  } else if (code == 0) {
-    monitorError("monitor:%p, save log failed, affect rows:%d", tsMonitorConn.conn, code);
-  } else {
-    monitorDebug("monitor:%p, save log info success, code:%s", tsMonitorConn.conn, tstrerror(code));
+    monitorDebug("monitor:%p, save %s info success, reason:%s", tsMonitorConn.conn, (char *)param, tstrerror(code));
   }
 }
 
@@ -359,7 +341,7 @@ static void monitorSaveSystemInfo() {
   pos += monitorBuildReqSql(sql + pos);
 
   monitorDebug("monitor:%p, save system info, sql:%s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertSysCallback, "log");
+  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sys");
 
   if (tsMonitorConn.timer != NULL && tsMonitorConn.state != MONITOR_STATE_STOPPED) {
     monitorStartTimer();
@@ -397,7 +379,7 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
           pMon->accessState);
 
   monitorDebug("monitor:%p, save account info, sql %s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertAcctCallback, "account");
+  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "account");
 }
 
 void monitorSaveLog(int32_t level, const char *const format, ...) {
@@ -421,14 +403,11 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
   sql[len++] = 0;
 
   monitorDebug("monitor:%p, save log, sql: %s", tsMonitorConn.conn, sql);
-  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorInsertLogCallback, "log");
+  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "log");
 }
 
 void monitorExecuteSQL(char *sql) {
   if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
-
   monitorDebug("monitor:%p, execute sql: %s", tsMonitorConn.conn, sql);
-  
-  // bug while insert binary
-  // taos_query_a(tsMonitorConn.conn, sql, NULL, NULL);
+  taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sql");
 }

src/plugins/mqtt/src/mqttSystem.c

@@ -64,7 +64,7 @@ int32_t mqttInitSystem() {
   }
 
   char* _begin_hostname = strstr(url, recntStatus.hostname);
-  if (strstr(_begin_hostname, ":") != NULL) {
+  if (_begin_hostname != NULL && strstr(_begin_hostname, ":") != NULL) {
     recntStatus.port = strbetween(_begin_hostname, ":", "/");
   } else {
     recntStatus.port = strbetween("'1883'", "'", "'");

src/query/inc/qast.h → src/query/inc/qAst.h

@@ -45,7 +45,6 @@ typedef void (*__do_filter_suppl_fn_t)(void *, void *);
  *
  */
 typedef struct tQueryInfo {
-  int32_t       colIndex; // index of column in schema
   uint8_t       optr;     // expression operator
   SSchema       sch;      // schema of tags
   char*         q;

src/query/inc/qExecutor.h

@@ -18,16 +18,16 @@
 #include "os.h"
 
 #include "hash.h"
-#include "qfill.h"
-#include "qresultBuf.h"
+#include "qFill.h"
+#include "qResultbuf.h"
+#include "qTsbuf.h"
 #include "qsqlparser.h"
-#include "qtsbuf.h"
+#include "query.h"
 #include "taosdef.h"
 #include "tarray.h"
 #include "tlockfree.h"
 #include "tsdb.h"
 #include "tsqlfunction.h"
-#include "query.h"
 
 struct SColumnFilterElem;
 typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, char* val1, char* val2);
@@ -158,7 +158,7 @@ typedef struct SQueryRuntimeEnv {
   SResultInfo*         resultInfo;       // todo refactor to merge with SWindowResInfo
   SQuery*              pQuery;
   SQLFunctionCtx*      pCtx;
-  int16_t              numOfRowsPerPage;
+  int32_t              numOfRowsPerPage;
   int16_t              offset[TSDB_MAX_COLUMNS];
   uint16_t             scanFlag;         // denotes reversed scan of data or not
   SFillInfo*           pFillInfo;

src/query/inc/qextbuffer.h → src/query/inc/qExtbuffer.h

@@ -28,7 +28,7 @@ extern "C" {
 #include "tdataformat.h"
 #include "talgo.h"
 
-#define DEFAULT_PAGE_SIZE (1024L*64)  // 16k larger than the SHistoInfo
+#define DEFAULT_PAGE_SIZE (1024L*4)  // 16k larger than the SHistoInfo
 #define MAX_TMPFILE_PATH_LENGTH PATH_MAX
 #define INITIAL_ALLOCATION_BUFFER_SIZE 64
 

src/query/inc/qfill.h → src/query/inc/qFill.h

@@ -21,8 +21,8 @@ extern "C" {
 #endif
 
 #include "os.h"
+#include "qExtbuffer.h"
 #include "taosdef.h"
-#include "qextbuffer.h"
 
 typedef struct {
   STColumn col;             // column info

src/query/inc/qpercentile.h → src/query/inc/qPercentile.h

@@ -16,7 +16,7 @@
 #ifndef TDENGINE_QPERCENTILE_H
 #define TDENGINE_QPERCENTILE_H
 
-#include "qextbuffer.h"
+#include "qExtbuffer.h"
 
 typedef struct MinMaxEntry {
   union {

src/query/inc/qresultBuf.h → src/query/inc/qResultbuf.h

@@ -20,9 +20,9 @@
 extern "C" {
 #endif
 
-#include "os.h"
-#include "qextbuffer.h"
 #include "hash.h"
+#include "os.h"
+#include "qExtbuffer.h"
 
 typedef struct SArray* SIDList;
 
@@ -33,14 +33,20 @@ typedef struct SDiskbasedResultBuf {
   int32_t   fd;                  // data file fd
   int32_t   allocateId;          // allocated page id
   int32_t   incStep;             // minimum allocated pages
-  char*     pBuf;                // mmap buffer pointer
+  void*     pBuf;                // mmap buffer pointer
   char*     path;                // file path
-  
+  int32_t   pageSize;            // current used page size
+  int32_t   inMemPages;          // numOfPages that are allocated in memory
   SHashObj* idsTable;            // id hash table
   SIDList   list;                // for each id, there is a page id list
+
+  void*     iBuf;                // inmemory buf
+  void*     handle;              // for debug purpose
+  void*     emptyDummyIdList;    // dummy id list
 } SDiskbasedResultBuf;
 
-#define DEFAULT_INTERN_BUF_PAGE_SIZE (8192L*5)
+#define DEFAULT_INTERN_BUF_PAGE_SIZE (1024L)
+#define DEFAULT_INMEM_BUF_PAGES       10
 
 /**
  * create disk-based result buffer
@@ -49,7 +55,8 @@ typedef struct SDiskbasedResultBuf {
  * @param rowSize
  * @return
  */
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t size, int32_t rowSize, void* handle);
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t numOfPages, int32_t rowSize, int32_t pagesize,
+    int32_t inMemPages, void* handle);
 
 /**
  *
@@ -81,8 +88,13 @@ SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId);
  * @param id
  * @return
  */
-#define GET_RES_BUF_PAGE_BY_ID(buf, id)  ((tFilePage*)((buf)->pBuf + DEFAULT_INTERN_BUF_PAGE_SIZE*(id)))
-
+static FORCE_INLINE tFilePage* getResBufPage(SDiskbasedResultBuf* pResultBuf, int32_t id) {
+  if (id < pResultBuf->inMemPages) {
+    return (tFilePage*) ((char*) pResultBuf->iBuf + id * pResultBuf->pageSize);
+  } else {
+    return (tFilePage*) ((char*) pResultBuf->pBuf + (id - pResultBuf->inMemPages) * pResultBuf->pageSize);
+  }
+}
 /**
  * get the total buffer size in the format of disk file
  * @param pResultBuf

src/query/inc/qUtil.h

@@ -49,7 +49,7 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
   assert(pResult != NULL && pRuntimeEnv != NULL);
 
   SQuery    *pQuery = pRuntimeEnv->pQuery;
-  tFilePage *page = GET_RES_BUF_PAGE_BY_ID(pRuntimeEnv->pResultBuf, pResult->pos.pageId);
+  tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pResult->pos.pageId);
   int32_t realRowId = pResult->pos.rowId * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pRuntimeEnv->topBotQuery, pRuntimeEnv->stableQuery);
 
   return ((char *)page->data) + pRuntimeEnv->offset[columnIndex] * pRuntimeEnv->numOfRowsPerPage +
@@ -59,6 +59,4 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
 __filter_func_t *getRangeFilterFuncArray(int32_t type);
 __filter_func_t *getValueFilterFuncArray(int32_t type);
 
-bool supportPrefilter(int32_t type);
-
 #endif  // TDENGINE_QUERYUTIL_H

src/query/inc/tsqlfunction.h

@@ -255,7 +255,15 @@ extern int32_t functionCompatList[]; // compatible check array list
 
 bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const char *minval, const char *maxval);
 
-void resetResultInfo(SResultInfo *pResInfo);
+/**
+ * the numOfRes should be kept, since it may be used later
+ * and allow the ResultInfo to be re initialized
+ */
+#define RESET_RESULT_INFO(_r)  \
+  do {                         \
+    (_r)->initialized = false; \
+  } while (0)
+
 void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf);
 
 static FORCE_INLINE void initResultInfo(SResultInfo *pResInfo) {

src/query/src/qast.c → src/query/src/qAst.c

@@ -16,17 +16,17 @@
 
 #include "os.h"
 
-#include "tname.h"
-#include "qast.h"
-#include "tsdb.h"
 #include "exception.h"
+#include "qAst.h"
+#include "qSyntaxtreefunction.h"
 #include "qsqlparser.h"
-#include "qsyntaxtreefunction.h"
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "tarray.h"
 #include "tbuffer.h"
 #include "tcompare.h"
+#include "tname.h"
+#include "tsdb.h"
 #include "tskiplist.h"
 #include "tsqlfunction.h"
 #include "tstoken.h"
@@ -678,7 +678,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
 
     tstr *name = (tstr*) tsdbGetTableName(*(void**) pData);
     // todo speed up by using hash
-    if (pQueryInfo->colIndex == TSDB_TBNAME_COLUMN_INDEX) {
+    if (pQueryInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
       if (pQueryInfo->optr == TSDB_RELATION_IN) {
         addToResult = pQueryInfo->compare(name, pQueryInfo->q);
       } else if (pQueryInfo->optr == TSDB_RELATION_LIKE) {
@@ -716,7 +716,7 @@ void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, S
     }
 
     tQueryInfo *pQueryInfo = pExpr->_node.info;
-    if (pQueryInfo->colIndex == 0 && pQueryInfo->optr != TSDB_RELATION_LIKE) {
+    if (pQueryInfo->sch.colId == PRIMARYKEY_TIMESTAMP_COL_INDEX && pQueryInfo->optr != TSDB_RELATION_LIKE) {
       tQueryIndexColumn(pSkipList, pQueryInfo, result);
     } else {
       tQueryIndexlessColumn(pSkipList, pQueryInfo, result, param->nodeFilterFn);

src/query/src/qextbuffer.c → src/query/src/qExtbuffer.c

@@ -12,16 +12,15 @@
  * 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 "qExtbuffer.h"
 #include "os.h"
-#include "tulog.h"
-#include "qextbuffer.h"
+#include "queryLog.h"
 #include "taos.h"
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "tsqlfunction.h"
-#include "ttime.h"
+#include "tulog.h"
 #include "tutil.h"
-#include "queryLog.h"
 
 #define COLMODEL_GET_VAL(data, schema, allrow, rowId, colId) \
   (data + (schema)->pFields[colId].offset * (allrow) + (rowId) * (schema)->pFields[colId].field.bytes)

src/query/src/qfill.c → src/query/src/qFill.c

@@ -13,9 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "qFill.h"
 #include "os.h"
-#include "qfill.h"
-#include "qextbuffer.h"
+#include "qExtbuffer.h"
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "tsqlfunction.h"

src/query/src/qFilterFunc.c → src/query/src/qFilterfunc.c

@@ -554,5 +554,3 @@ __filter_func_t* getValueFilterFuncArray(int32_t type) {
     default: return NULL;
   }
 }
-
-bool supportPrefilter(int32_t type) { return type != TSDB_DATA_TYPE_BINARY && type != TSDB_DATA_TYPE_NCHAR; }

src/query/src/qhistogram.c → src/query/src/qHistogram.c

@@ -14,7 +14,7 @@
  */
 #include "os.h"
 
-#include "qhistogram.h"
+#include "qHistogram.h"
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "tlosertree.h"

src/query/src/qpercentile.c → src/query/src/qPercentile.c

@@ -13,12 +13,12 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "qPercentile.h"
 #include "os.h"
-#include "tulog.h"
-#include "qpercentile.h"
+#include "queryLog.h"
 #include "taosdef.h"
 #include "taosmsg.h"
-#include "queryLog.h"
+#include "tulog.h"
 
 tExtMemBuffer *releaseBucketsExceptFor(tMemBucket *pMemBucket, int16_t segIdx, int16_t slotIdx) {
   tExtMemBuffer *pBuffer = NULL;

src/query/src/qresultBuf.c → src/query/src/qResultbuf.c

-#include "qresultBuf.h"
+#include "qResultbuf.h"
 #include "hash.h"
-#include "qextbuffer.h"
-#include "taoserror.h"
+#include "qExtbuffer.h"
 #include "queryLog.h"
+#include "taoserror.h"
+
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t numOfPages, int32_t rowSize,
+    int32_t pagesize, int32_t inMemPages, void* handle) {
 
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t size, int32_t rowSize, void* handle) {
   *pResultBuf = calloc(1, sizeof(SDiskbasedResultBuf));
   SDiskbasedResultBuf* pResBuf = *pResultBuf;
   if (pResBuf == NULL) {
     return TSDB_CODE_COM_OUT_OF_MEMORY;  
   }
-  
-  pResBuf->numOfRowsPerPage = (DEFAULT_INTERN_BUF_PAGE_SIZE - sizeof(tFilePage)) / rowSize;
-  pResBuf->numOfPages = size;
 
-  pResBuf->totalBufSize = pResBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE;
+  pResBuf->pageSize   = pagesize;
+  pResBuf->numOfPages = inMemPages;  // all pages are in buffer in the first place
+  pResBuf->inMemPages = inMemPages;
+  assert(inMemPages <= numOfPages);
+
+  pResBuf->numOfRowsPerPage = (pagesize - sizeof(tFilePage)) / rowSize;
+
+  pResBuf->totalBufSize = pResBuf->numOfPages * pagesize;
   pResBuf->incStep = 4;
+  pResBuf->allocateId = -1;
+
+  pResBuf->iBuf = calloc(pResBuf->inMemPages, pResBuf->pageSize);
 
   // init id hash table
-  pResBuf->idsTable = taosHashInit(size, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false);
-  pResBuf->list = taosArrayInit(size, POINTER_BYTES);
+  pResBuf->idsTable = taosHashInit(numOfPages, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false);
+  pResBuf->list = taosArrayInit(numOfPages, POINTER_BYTES);
 
-  char path[4096] = {0};
+  char path[PATH_MAX] = {0};
   getTmpfilePath("tsdb_qbuf", path);
   pResBuf->path = strdup(path);
 
-  pResBuf->fd = open(pResBuf->path, O_CREAT | O_RDWR, 0666);
-  if (!FD_VALID(pResBuf->fd)) {
-    qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno));
+  pResBuf->fd = FD_INITIALIZER;
+  pResBuf->pBuf = NULL;
+  pResBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
+
+  qDebug("QInfo:%p create resBuf for output, page size:%d, initial pages:%d, %" PRId64 "bytes", handle,
+      pResBuf->pageSize, pResBuf->numOfPages, pResBuf->totalBufSize);
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t getNumOfResultBufGroupId(SDiskbasedResultBuf* pResultBuf) { return taosHashGetSize(pResultBuf->idsTable); }
+
+int32_t getResBufSize(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->totalBufSize; }
+
+#define NUM_OF_PAGES_ON_DISK(_r) ((_r)->numOfPages - (_r)->inMemPages)
+#define FILE_SIZE_ON_DISK(_r) (NUM_OF_PAGES_ON_DISK(_r) * (_r)->pageSize)
+
+static int32_t createDiskResidesBuf(SDiskbasedResultBuf* pResultBuf) {
+  pResultBuf->fd = open(pResultBuf->path, O_CREAT | O_RDWR, 0666);
+  if (!FD_VALID(pResultBuf->fd)) {
+    qError("failed to create tmp file: %s on disk. %s", pResultBuf->path, strerror(errno));
     return TAOS_SYSTEM_ERROR(errno);
   }
 
-  int32_t ret = ftruncate(pResBuf->fd, pResBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE);
+  assert(pResultBuf->numOfPages == pResultBuf->inMemPages);
+  pResultBuf->numOfPages += pResultBuf->incStep;
+
+  int32_t ret = ftruncate(pResultBuf->fd, NUM_OF_PAGES_ON_DISK(pResultBuf) * pResultBuf->pageSize);
   if (ret != TSDB_CODE_SUCCESS) {
-    qError("failed to create tmp file: %s on disk. %s", pResBuf->path, strerror(errno));
+    qError("failed to create tmp file: %s on disk. %s", pResultBuf->path, strerror(errno));
     return TAOS_SYSTEM_ERROR(errno);
   }
 
-  pResBuf->pBuf = mmap(NULL, pResBuf->totalBufSize, PROT_READ | PROT_WRITE, MAP_SHARED, pResBuf->fd, 0);
-  if (pResBuf->pBuf == MAP_FAILED) {
-    qError("QInfo:%p failed to map temp file: %s. %s", handle, pResBuf->path, strerror(errno));
+  pResultBuf->pBuf = mmap(NULL, FILE_SIZE_ON_DISK(pResultBuf), PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0);
+  if (pResultBuf->pBuf == MAP_FAILED) {
+    qError("QInfo:%p failed to map temp file: %s. %s", pResultBuf->handle, pResultBuf->path, strerror(errno));
     return TAOS_SYSTEM_ERROR(errno);
   }
 
-  qDebug("QInfo:%p create tmp file for output result:%s, %" PRId64 "bytes", handle, pResBuf->path,
-      pResBuf->totalBufSize);
-  
+  pResultBuf->totalBufSize = pResultBuf->numOfPages * pResultBuf->pageSize;
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t getNumOfResultBufGroupId(SDiskbasedResultBuf* pResultBuf) { return taosHashGetSize(pResultBuf->idsTable); }
+static int32_t extendDiskFileSize(SDiskbasedResultBuf* pResultBuf, int32_t incNumOfPages) {
+  assert(pResultBuf->numOfPages * pResultBuf->pageSize == pResultBuf->totalBufSize);
+  int32_t ret = TSDB_CODE_SUCCESS;
 
-int32_t getResBufSize(SDiskbasedResultBuf* pResultBuf) { return pResultBuf->totalBufSize; }
+  if (pResultBuf->pBuf == NULL) {
+    assert(pResultBuf->fd == FD_INITIALIZER);
 
-static int32_t extendDiskFileSize(SDiskbasedResultBuf* pResultBuf, int32_t numOfPages) {
-  assert(pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE == pResultBuf->totalBufSize);
-
-  int32_t ret = munmap(pResultBuf->pBuf, pResultBuf->totalBufSize);
-  pResultBuf->numOfPages += numOfPages;
-
-  /*
-   * disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may
-   * be insufficient
-   */
-  ret = ftruncate(pResultBuf->fd, pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE);
-  if (ret != 0) {
-    //    dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile,
-    //           strerror(errno));
-    return TSDB_CODE_QRY_NO_DISKSPACE;
-  }
+    if ((ret = createDiskResidesBuf(pResultBuf)) != TSDB_CODE_SUCCESS) {
+      return ret;
+    }
+  } else {
+    ret = munmap(pResultBuf->pBuf, FILE_SIZE_ON_DISK(pResultBuf));
+    pResultBuf->numOfPages += incNumOfPages;
+
+    /*
+     * disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may
+     * be insufficient
+     */
+    ret = ftruncate(pResultBuf->fd, NUM_OF_PAGES_ON_DISK(pResultBuf) * pResultBuf->pageSize);
+    if (ret != TSDB_CODE_SUCCESS) {
+      //    dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile,
+      //           strerror(errno));
+      return TSDB_CODE_QRY_NO_DISKSPACE;
+    }
 
-  pResultBuf->totalBufSize = pResultBuf->numOfPages * DEFAULT_INTERN_BUF_PAGE_SIZE;
-  pResultBuf->pBuf = mmap(NULL, pResultBuf->totalBufSize, PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0);
+    pResultBuf->totalBufSize = pResultBuf->numOfPages * pResultBuf->pageSize;
+    pResultBuf->pBuf = mmap(NULL, FILE_SIZE_ON_DISK(pResultBuf), PROT_READ | PROT_WRITE, MAP_SHARED, pResultBuf->fd, 0);
 
-  if (pResultBuf->pBuf == MAP_FAILED) {
-    //    dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno));
-    return TSDB_CODE_QRY_OUT_OF_MEMORY;
+    if (pResultBuf->pBuf == MAP_FAILED) {
+      //    dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno));
+      return TSDB_CODE_QRY_OUT_OF_MEMORY;
+    }
   }
 
   return TSDB_CODE_SUCCESS;
 }
 
-static FORCE_INLINE bool noMoreAvailablePages(SDiskbasedResultBuf* pResultBuf) {
-  return (pResultBuf->allocateId == pResultBuf->numOfPages - 1);
-}
+#define NO_AVAILABLE_PAGES(_b) ((_b)->allocateId == (_b)->numOfPages - 1)
 
 static FORCE_INLINE int32_t getGroupIndex(SDiskbasedResultBuf* pResultBuf, int32_t groupId) {
   assert(pResultBuf != NULL);
@@ -121,20 +152,19 @@ static void registerPageId(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int
 }
 
 tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t* pageId) {
-  if (noMoreAvailablePages(pResultBuf)) {
+  if (NO_AVAILABLE_PAGES(pResultBuf)) {
     if (extendDiskFileSize(pResultBuf, pResultBuf->incStep) != TSDB_CODE_SUCCESS) {
       return NULL;
     }
   }
 
   // register new id in this group
-  *pageId = (pResultBuf->allocateId++);
+  *pageId = (++pResultBuf->allocateId);
   registerPageId(pResultBuf, groupId, *pageId);
 
-  tFilePage* page = GET_RES_BUF_PAGE_BY_ID(pResultBuf, *pageId);
-  
   // clear memory for the new page
-  memset(page, 0, DEFAULT_INTERN_BUF_PAGE_SIZE);
+  tFilePage* page = getResBufPage(pResultBuf, *pageId);
+  memset(page, 0, pResultBuf->pageSize);
   
   return page;
 }
@@ -144,7 +174,7 @@ int32_t getNumOfRowsPerPage(SDiskbasedResultBuf* pResultBuf) { return pResultBuf
 SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId) {
   int32_t slot = getGroupIndex(pResultBuf, groupId);
   if (slot < 0) {
-    return taosArrayInit(1, sizeof(int32_t));
+    return pResultBuf->emptyDummyIdList;
   } else {
     return taosArrayGetP(pResultBuf->list, slot);
   }
@@ -156,13 +186,18 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) {
   }
 
   if (FD_VALID(pResultBuf->fd)) {
+    qDebug("QInfo:%p disk-based output buffer closed, total:%" PRId64 " bytes, file created:%s, file size:%d", handle,
+        pResultBuf->totalBufSize, pResultBuf->path, FILE_SIZE_ON_DISK(pResultBuf));
+
     close(pResultBuf->fd);
+    munmap(pResultBuf->pBuf, FILE_SIZE_ON_DISK(pResultBuf));
+    pResultBuf->pBuf = NULL;
+  } else {
+    qDebug("QInfo:%p disk-based output buffer closed, total:%" PRId64 " bytes, no file created", handle,
+           pResultBuf->totalBufSize);
   }
 
-  qDebug("QInfo:%p disk-based output buffer closed, %" PRId64 " bytes, file:%s", handle, pResultBuf->totalBufSize, pResultBuf->path);
-  munmap(pResultBuf->pBuf, pResultBuf->totalBufSize);
   unlink(pResultBuf->path);
-  
   tfree(pResultBuf->path);
 
   size_t size = taosArrayGetSize(pResultBuf->list);
@@ -172,8 +207,10 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) {
   }
 
   taosArrayDestroy(pResultBuf->list);
+  taosArrayDestroy(pResultBuf->emptyDummyIdList);
   taosHashCleanup(pResultBuf->idsTable);
-  
+
+  tfree(pResultBuf->iBuf);
   tfree(pResultBuf);
 }
 

src/query/src/qsyntaxtreefunction.c → src/query/src/qSyntaxtreefunction.c

@@ -15,7 +15,7 @@
 
 #include "os.h"
 
-#include "qsyntaxtreefunction.h"
+#include "qSyntaxtreefunction.h"
 #include "taosdef.h"
 #include "tutil.h"
 

src/query/src/qtsbuf.c → src/query/src/qTsbuf.c

-#include "qtsbuf.h"
+#include "qTsbuf.h"
+#include "taoserror.h"
 #include "tscompression.h"
 #include "tutil.h"
-#include "taoserror.h"
 
 static int32_t getDataStartOffset();
 static void TSBufUpdateVnodeInfo(STSBuf* pTSBuf, int32_t index, STSVnodeBlockInfo* pBlockInfo);

src/query/src/qUtil.c

@@ -26,12 +26,10 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
   int32_t size = 0;
 
   for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
-    assert(pQuery->pSelectExpr[i].interBytes <= DEFAULT_INTERN_BUF_PAGE_SIZE);
     size += pQuery->pSelectExpr[i].interBytes;
   }
 
   assert(size > 0);
-
   return size;
 }
 
@@ -243,7 +241,7 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
     size_t size = pRuntimeEnv->pQuery->pSelectExpr[i].bytes;
     memset(s, 0, size);
     
-    resetResultInfo(pResultInfo);
+    RESET_RESULT_INFO(pResultInfo);
   }
   
   pWindowRes->numOfRows = 0;

src/query/tests/CMakeLists.txt

@@ -11,5 +11,5 @@ IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
     AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 
     ADD_EXECUTABLE(queryTest ${SOURCE_LIST})
-    TARGET_LINK_LIBRARIES(queryTest taos query gtest pthread)
-ENDIF()
+    TARGET_LINK_LIBRARIES(queryTest taos query gtest pthread gcov)
+ENDIF()
\ No newline at end of file

src/query/tests/astTest.cpp

@@ -3,8 +3,8 @@
 #include <cassert>
 #include <iostream>
 
+#include "qAst.h"
 #include "taosmsg.h"
-#include "qast.h"
 #include "tsdb.h"
 #include "tskiplist.h"
 

src/query/tests/histogramTest.cpp

@@ -9,7 +9,7 @@
 #include "tstoken.h"
 #include "tutil.h"
 
-#include "qhistogram.h"
+#include "qHistogram.h"
 
 /* test validate the names for table/database */
 TEST(testCase, histogram_binary_search) {

src/query/tests/resultBufferTest.cpp

@@ -2,15 +2,15 @@
 #include <cassert>
 #include <iostream>
 
+#include "qResultbuf.h"
 #include "taos.h"
-#include "qresultBuf.h"
 #include "tsdb.h"
 
 namespace {
 // simple test
 void simpleTest() {
   SDiskbasedResultBuf* pResultBuf = NULL;
-  int32_t ret = createDiskbasedResultBuffer(&pResultBuf, 1000, 64, NULL);
+  int32_t ret = createDiskbasedResultBuffer(&pResultBuf, 1000, 64, 1024, 4, NULL);
   
   int32_t pageId = 0;
   int32_t groupId = 0;
@@ -22,8 +22,7 @@ void simpleTest() {
   ASSERT_EQ(getResBufSize(pResultBuf), 1000*16384L);
   
   SIDList list = getDataBufPagesIdList(pResultBuf, groupId);
-  ASSERT_EQ(list.size, 1);
-  
+  ASSERT_EQ(taosArrayGetSize(list), 1);
   ASSERT_EQ(getNumOfResultBufGroupId(pResultBuf), 1);
   
   destroyResultBuf(pResultBuf, NULL);

src/query/tests/tsBufTest.cpp

@@ -5,10 +5,10 @@
 #include "taos.h"
 #include "tsdb.h"
 
+#include "qTsbuf.h"
 #include "tstoken.h"
 #include "ttime.h"
 #include "tutil.h"
-#include "qtsbuf.h"
 
 namespace {
 /**

src/rpc/src/rpcMain.c

@@ -446,7 +446,10 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
   // set the idle timer to monitor the activity
   taosTmrReset(rpcProcessIdleTimer, pRpc->idleTime, pConn, pRpc->tmrCtrl, &pConn->pIdleTimer);
   rpcSendMsgToPeer(pConn, msg, msgLen);
-  pConn->secured = 1; // connection shall be secured
+
+  // if not set to secured, set it expcet NOT_READY case, since client wont treat it as secured
+  if (pConn->secured == 0 && pMsg->code != TSDB_CODE_RPC_NOT_READY)
+    pConn->secured = 1; // connection shall be secured
 
   if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg);
   pConn->pReqMsg = NULL;

src/tsdb/inc/tsdbMain.h

@@ -42,6 +42,7 @@ extern int tsdbDebugFlag;
 #define TSDB_MAX_TABLE_SCHEMAS 16
 #define TSDB_FILE_HEAD_SIZE 512
 #define TSDB_FILE_DELIMITER 0xF00AFA0F
+#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
 
 // Definitions
 // ------------------ tsdbMeta.c
@@ -132,21 +133,30 @@ typedef struct {
 // ------------------ tsdbFile.c
 extern const char* tsdbFileSuffix[];
 typedef enum {
+#ifdef TSDB_IDX
+  TSDB_FILE_TYPE_IDX = 0,
+  TSDB_FILE_TYPE_HEAD,
+#else
   TSDB_FILE_TYPE_HEAD = 0,
+#endif
   TSDB_FILE_TYPE_DATA,
   TSDB_FILE_TYPE_LAST,
   TSDB_FILE_TYPE_MAX,
+#ifdef TSDB_IDX
+  TSDB_FILE_TYPE_NIDX,
+#endif
   TSDB_FILE_TYPE_NHEAD,
   TSDB_FILE_TYPE_NLAST
 } TSDB_FILE_TYPE;
 
 typedef struct {
-  uint32_t offset;
+  uint32_t magic;
   uint32_t len;
-  uint64_t size;      // total size of the file
-  uint64_t tombSize;  // unused file size
   uint32_t totalBlocks;
   uint32_t totalSubBlocks;
+  uint32_t offset;
+  uint64_t size;      // total size of the file
+  uint64_t tombSize;  // unused file size
 } STsdbFileInfo;
 
 typedef struct {
@@ -197,6 +207,7 @@ typedef struct {
 
 // ------------------ tsdbRWHelper.c
 typedef struct {
+  int32_t  tid;
   uint32_t len;
   uint32_t offset;
   uint32_t hasLast : 2;
@@ -220,7 +231,7 @@ typedef struct {
 
 typedef struct {
   int32_t    delimiter;  // For recovery usage
-  int32_t    checksum;   // TODO: decide if checksum logic in this file or make it one API
+  int32_t    tid;
   uint64_t   uid;
   SCompBlock blocks[];
 } SCompInfo;
@@ -249,24 +260,27 @@ typedef struct {
 typedef enum { TSDB_WRITE_HELPER, TSDB_READ_HELPER } tsdb_rw_helper_t;
 
 typedef struct {
-  int   fid;
-  TSKEY minKey;
-  TSKEY maxKey;
-  // For read/write purpose
-  SFile headF;
-  SFile dataF;
-  SFile lastF;
-  // For write purpose only
-  SFile nHeadF;
-  SFile nLastF;
+  TSKEY      minKey;
+  TSKEY      maxKey;
+  SFileGroup fGroup;
+#ifdef TSDB_IDX
+  SFile nIdxF;
+#endif
+  SFile      nHeadF;
+  SFile      nLastF;
 } SHelperFile;
 
 typedef struct {
   uint64_t uid;
   int32_t  tid;
-  int32_t  sversion;
 } SHelperTable;
 
+typedef struct {
+  SCompIdx* pIdxArray;
+  int       numOfIdx;
+  int       curIdx;
+} SIdxH;
+
 typedef struct {
   tsdb_rw_helper_t type;
 
@@ -274,7 +288,9 @@ typedef struct {
   int8_t     state;
   // For file set usage
   SHelperFile files;
-  SCompIdx*   pCompIdx;
+  SIdxH       idxH;
+  SCompIdx    curCompIdx;
+  void*       pWIdx;
   // For table set usage
   SHelperTable tableInfo;
   SCompInfo*   pCompInfo;
@@ -286,7 +302,6 @@ typedef struct {
   void*      compBuffer;  // Buffer for temperary compress/decompress purpose
 } SRWHelper;
 
-
 // Operations
 // ------------------ tsdbMeta.c
 #define TABLE_TYPE(t) (t)->type
@@ -296,6 +311,7 @@ typedef struct {
 #define TABLE_TID(t) (t)->tableId.tid
 #define TABLE_SUID(t) (t)->suid
 #define TABLE_LASTKEY(t) (t)->lastKey
+#define TSDB_META_FILE_MAGIC(m) KVSTORE_MAGIC((m)->pStore)
 
 STsdbMeta* tsdbNewMeta(STsdbCfg* pCfg);
 void       tsdbFreeMeta(STsdbMeta* pMeta);
@@ -426,6 +442,7 @@ int         tsdbUpdateFileHeader(SFile* pFile, uint32_t version);
 int         tsdbEncodeSFileInfo(void** buf, const STsdbFileInfo* pInfo);
 void*       tsdbDecodeSFileInfo(void* buf, STsdbFileInfo* pInfo);
 void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
+void        tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
 
 // ------------------ tsdbRWHelper.c
 #define TSDB_HELPER_CLEAR_STATE 0x0        // Clear state
@@ -444,6 +461,16 @@ void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
 #define helperRepo(h) (h)->pRepo
 #define helperState(h) (h)->state
 #define TSDB_NLAST_FILE_OPENED(h) ((h)->files.nLastF.fd > 0)
+#define helperFileId(h) ((h)->files.fGroup.fileId)
+#ifdef TSDB_IDX
+#define helperIdxF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_IDX]))
+#define helperNewIdxF(h) (&((h)->files.nIdxF))
+#endif
+#define helperHeadF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_HEAD]))
+#define helperDataF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_DATA]))
+#define helperLastF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_LAST]))
+#define helperNewHeadF(h) (&((h)->files.nHeadF))
+#define helperNewLastF(h) (&((h)->files.nLastF))
 
 int  tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
 int  tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);

src/tsdb/src/tsdbFile.c

@@ -30,7 +30,11 @@
 #include "ttime.h"
 #include "tfile.h"
 
+#ifdef TSDB_IDX
+const char *tsdbFileSuffix[] = {".idx", ".head", ".data", ".last", "", ".i", ".h", ".l"};
+#else
 const char *tsdbFileSuffix[] = {".head", ".data", ".last", "", ".h", ".l"};
+#endif
 
 static int   tsdbInitFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type);
 static void  tsdbDestroyFile(SFile *pFile);
@@ -108,7 +112,7 @@ int tsdbOpenFileH(STsdbRepo *pRepo) {
 
     memset((void *)(&fileGroup), 0, sizeof(SFileGroup));
     fileGroup.fileId = fid;
-    for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
+    for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) {
       if (tsdbInitFile(&fileGroup.files[type], pRepo, fid, type) < 0) {
         tsdbError("vgId:%d failed to init file fid %d type %d", REPO_ID(pRepo), fid, type);
         goto _err;
@@ -126,7 +130,7 @@ int tsdbOpenFileH(STsdbRepo *pRepo) {
   return 0;
 
 _err:
-  for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) tsdbDestroyFile(&fileGroup.files[type]);
+  for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) tsdbDestroyFile(&fileGroup.files[type]);
 
   tfree(tDataDir);
   if (dir != NULL) closedir(dir);
@@ -139,7 +143,7 @@ void tsdbCloseFileH(STsdbRepo *pRepo) {
 
   for (int i = 0; i < pFileH->nFGroups; i++) {
     SFileGroup *pFGroup = pFileH->pFGroup + i;
-    for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
+    for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) {
       tsdbDestroyFile(&pFGroup->files[type]);
     }
   }
@@ -156,7 +160,7 @@ SFileGroup *tsdbCreateFGroupIfNeed(STsdbRepo *pRepo, char *dataDir, int fid, int
   SFileGroup *pGroup = tsdbSearchFGroup(pFileH, fid, TD_EQ);
   if (pGroup == NULL) {  // if not exists, create one
     pFGroup->fileId = fid;
-    for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
+    for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) {
       if (tsdbCreateFile(&pFGroup->files[type], pRepo, fid, type) < 0)
         goto _err;
     }
@@ -169,7 +173,7 @@ SFileGroup *tsdbCreateFGroupIfNeed(STsdbRepo *pRepo, char *dataDir, int fid, int
   return pGroup;
 
 _err:
-  for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) tsdbDestroyFile(&pGroup->files[type]);
+  for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) tsdbDestroyFile(&pGroup->files[type]);
   return NULL;
 }
 
@@ -260,6 +264,7 @@ int tsdbCreateFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type) {
   }
 
   pFile->info.size = TSDB_FILE_HEAD_SIZE;
+  pFile->info.magic = TSDB_FILE_INIT_MAGIC;
 
   if (tsdbUpdateFileHeader(pFile, 0) < 0) {
     tsdbCloseFile(pFile);
@@ -323,23 +328,25 @@ int tsdbUpdateFileHeader(SFile *pFile, uint32_t version) {
 
 int tsdbEncodeSFileInfo(void **buf, const STsdbFileInfo *pInfo) {
   int tlen = 0;
-  tlen += taosEncodeFixedU32(buf, pInfo->offset);
+  tlen += taosEncodeFixedU32(buf, pInfo->magic);
   tlen += taosEncodeFixedU32(buf, pInfo->len);
-  tlen += taosEncodeFixedU64(buf, pInfo->size);
-  tlen += taosEncodeFixedU64(buf, pInfo->tombSize);
   tlen += taosEncodeFixedU32(buf, pInfo->totalBlocks);
   tlen += taosEncodeFixedU32(buf, pInfo->totalSubBlocks);
+  tlen += taosEncodeFixedU32(buf, pInfo->offset);
+  tlen += taosEncodeFixedU64(buf, pInfo->size);
+  tlen += taosEncodeFixedU64(buf, pInfo->tombSize);
 
   return tlen;
 }
 
 void *tsdbDecodeSFileInfo(void *buf, STsdbFileInfo *pInfo) {
-  buf = taosDecodeFixedU32(buf, &(pInfo->offset));
+  buf = taosDecodeFixedU32(buf, &(pInfo->magic));
   buf = taosDecodeFixedU32(buf, &(pInfo->len));
-  buf = taosDecodeFixedU64(buf, &(pInfo->size));
-  buf = taosDecodeFixedU64(buf, &(pInfo->tombSize));
   buf = taosDecodeFixedU32(buf, &(pInfo->totalBlocks));
   buf = taosDecodeFixedU32(buf, &(pInfo->totalSubBlocks));
+  buf = taosDecodeFixedU32(buf, &(pInfo->offset));
+  buf = taosDecodeFixedU64(buf, &(pInfo->size));
+  buf = taosDecodeFixedU64(buf, &(pInfo->tombSize));
 
   return buf;
 }
@@ -358,7 +365,7 @@ void tsdbRemoveFileGroup(STsdbRepo *pRepo, SFileGroup *pFGroup) {
   pFileH->nFGroups--;
   ASSERT(pFileH->nFGroups >= 0);
 
-  for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
+  for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) {
     if (remove(fileGroup.files[type].fname) < 0) {
       tsdbError("vgId:%d failed to remove file %s", REPO_ID(pRepo), fileGroup.files[type].fname);
     }

src/tsdb/src/tsdbMain.c

@@ -212,59 +212,61 @@ uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_
 
   char *sdup = strdup(pRepo->rootDir);
   char *prefix = dirname(sdup);
+  int   prefixLen = strlen(prefix);
+  tfree(sdup);
 
   if (name[0] == 0) {  // get the file from index or after, but not larger than eindex
-    int fid = (*index) / 3;
+    int fid = (*index) / TSDB_FILE_TYPE_MAX;
 
     if (pFileH->nFGroups == 0 || fid > pFileH->pFGroup[pFileH->nFGroups - 1].fileId) {
       if (*index <= TSDB_META_FILE_INDEX && TSDB_META_FILE_INDEX <= eindex) {
         fname = tsdbGetMetaFileName(pRepo->rootDir);
         *index = TSDB_META_FILE_INDEX;
+        magic = TSDB_META_FILE_MAGIC(pRepo->tsdbMeta);
       } else {
-        tfree(sdup);
         return 0;
       }
     } else {
       SFileGroup *pFGroup =
           taosbsearch(&fid, pFileH->pFGroup, pFileH->nFGroups, sizeof(SFileGroup), keyFGroupCompFunc, TD_GE);
       if (pFGroup->fileId == fid) {
-        fname = strdup(pFGroup->files[(*index) % 3].fname);
+        fname = strdup(pFGroup->files[(*index) % TSDB_FILE_TYPE_MAX].fname);
+        magic = pFGroup->files[(*index) % TSDB_FILE_TYPE_MAX].info.magic;
       } else {
-        if (pFGroup->fileId * 3 + 2 < eindex) {
+        if ((pFGroup->fileId + 1) * TSDB_FILE_TYPE_MAX - 1 < eindex) {
           fname = strdup(pFGroup->files[0].fname);
-          *index = pFGroup->fileId * 3;
+          *index = pFGroup->fileId * TSDB_FILE_TYPE_MAX;
+          magic = pFGroup->files[0].info.magic;
         } else {
-          tfree(sdup);
           return 0;
         }
       }
     }
-    strcpy(name, fname + strlen(prefix));
+    strcpy(name, fname + prefixLen);
   } else {                                 // get the named file at the specified index. If not there, return 0
     if (*index == TSDB_META_FILE_INDEX) {  // get meta file
       fname = tsdbGetMetaFileName(pRepo->rootDir);
+      magic = TSDB_META_FILE_MAGIC(pRepo->tsdbMeta);
     } else {
-      int         fid = (*index) / 3;
+      int         fid = (*index) / TSDB_FILE_TYPE_MAX;
       SFileGroup *pFGroup = tsdbSearchFGroup(pFileH, fid, TD_EQ);
       if (pFGroup == NULL) {  // not found
-        tfree(sdup);
         return 0;
       }
 
-      SFile *pFile = &pFGroup->files[(*index) % 3];
+      SFile *pFile = &pFGroup->files[(*index) % TSDB_FILE_TYPE_MAX];
       fname = strdup(pFile->fname);
+      magic = pFile->info.magic;
     }
   }
 
   if (stat(fname, &fState) < 0) {
-    tfree(sdup);
     tfree(fname);
     return 0;
   }
 
-  tfree(sdup);
   *size = fState.st_size;
-  magic = *size;
+  // magic = *size;
 
   tfree(fname);
   return magic;
@@ -793,7 +795,8 @@ static int tsdbRestoreInfo(STsdbRepo *pRepo) {
     for (int i = 1; i < pRepo->config.maxTables; i++) {
       STable *pTable = pMeta->tables[i];
       if (pTable == NULL) continue;
-      SCompIdx *pIdx = &rhelper.pCompIdx[i];
+      tsdbSetHelperTable(&rhelper, pTable, pRepo);
+      SCompIdx *pIdx = &(rhelper.curCompIdx);
 
       if (pIdx->offset > 0 && pTable->lastKey < pIdx->maxKey) pTable->lastKey = pIdx->maxKey;
     }

src/tsdb/src/tsdbMemTable.c

@@ -31,7 +31,6 @@ static int         tsdbCommitMeta(STsdbRepo *pRepo);
 static void        tsdbEndCommit(STsdbRepo *pRepo);
 static int         tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSKEY maxKey);
 static int  tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHelper *pHelper, SDataCols *pDataCols);
-static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
 static SCommitIter *tsdbCreateCommitIters(STsdbRepo *pRepo);
 static void         tsdbDestroyCommitIters(SCommitIter *iters, int maxTables);
 
@@ -265,13 +264,11 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
   }
 
   do {
-    if (numOfRows >= maxRowsToRead) break;
-
     SDataRow row = tsdbNextIterRow(pIter);
     if (row == NULL) break;
 
     keyNext = dataRowKey(row);
-    if (keyNext < 0 || keyNext > maxKey) break;
+    if (keyNext > maxKey) break;
 
     bool keyFiltered = false;
     if (nFilterKeys != 0) {
@@ -290,6 +287,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
     }
 
     if (!keyFiltered) {
+      if (numOfRows >= maxRowsToRead) break;
       if (pCols) {
         if (pSchema == NULL || schemaVersion(pSchema) != dataRowVersion(row)) {
           pSchema = tsdbGetTableSchemaImpl(pTable, false, false, dataRowVersion(row));
@@ -544,7 +542,7 @@ static int tsdbHasDataToCommit(SCommitIter *iters, int nIters, TSKEY minKey, TSK
   return 0;
 }
 
-static void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey) {
+void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey) {
   *minKey = fileId * daysPerFile * tsMsPerDay[precision];
   *maxKey = *minKey + daysPerFile * tsMsPerDay[precision] - 1;
 }
@@ -628,9 +626,12 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
   tsdbCloseHelperFile(pHelper, 0);
 
   pthread_rwlock_wrlock(&(pFileH->fhlock));
-  pGroup->files[TSDB_FILE_TYPE_HEAD] = pHelper->files.headF;
-  pGroup->files[TSDB_FILE_TYPE_DATA] = pHelper->files.dataF;
-  pGroup->files[TSDB_FILE_TYPE_LAST] = pHelper->files.lastF;
+#ifdef TSDB_IDX
+  pGroup->files[TSDB_FILE_TYPE_IDX] = *(helperIdxF(pHelper));
+#endif
+  pGroup->files[TSDB_FILE_TYPE_HEAD] = *(helperHeadF(pHelper));
+  pGroup->files[TSDB_FILE_TYPE_DATA] = *(helperDataF(pHelper));
+  pGroup->files[TSDB_FILE_TYPE_LAST] = *(helperLastF(pHelper));
   pthread_rwlock_unlock(&(pFileH->fhlock));
 
   return 0;

src/tsdb/src/tsdbMeta.c

@@ -123,7 +123,10 @@ int tsdbCreateTable(TSDB_REPO_T *repo, STableCfg *pCfg) {
   int   tlen2 = tsdbGetTableEncodeSize(TSDB_UPDATE_META, table);
   int   tlen = tlen1 + tlen2;
   void *buf = tsdbAllocBytes(pRepo, tlen);
-  ASSERT(buf != NULL);
+  if (buf == NULL) {
+    goto _err;
+  }
+
   if (newSuper) {
     void *pBuf = tsdbInsertTableAct(pRepo, TSDB_UPDATE_META, buf, super);
     ASSERT(POINTER_DISTANCE(pBuf, buf) == tlen1);

src/util/inc/tkvstore.h

@@ -25,10 +25,11 @@ typedef int (*iterFunc)(void *, void *cont, int contLen);
 typedef void (*afterFunc)(void *);
 
 typedef struct {
-  int64_t size;  // including 512 bytes of header size
-  int64_t tombSize;
-  int64_t nRecords;
-  int64_t nDels;
+  int64_t  size;  // including 512 bytes of header size
+  int64_t  tombSize;
+  int64_t  nRecords;
+  int64_t  nDels;
+  uint32_t magic;
 } SStoreInfo;
 
 typedef struct {
@@ -45,6 +46,8 @@ typedef struct {
   SStoreInfo info;
 } SKVStore;
 
+#define KVSTORE_MAGIC(s) (s)->info.magic
+
 int       tdCreateKVStore(char *fname);
 int       tdDestroyKVStore(char *fname);
 SKVStore *tdOpenKVStore(char *fname, iterFunc iFunc, afterFunc aFunc, void *appH);

src/util/inc/tutil.h

@@ -35,12 +35,12 @@ extern "C" {
 #define WCHAR wchar_t
 
 #define tfree(x)         \
-  {                      \
+  do {                   \
     if (x) {             \
       free((void *)(x)); \
       x = 0;             \
     }                    \
-  }
+  } while(0);
 
 #define tstrncpy(dst, src, size)   \
   do {                             \

src/util/src/tconfig.c

@@ -308,38 +308,47 @@ bool taosReadGlobalCfg() {
 
   sprintf(fileName, "%s/taos.cfg", configDir);
   FILE* fp = fopen(fileName, "r");
+  if (fp == NULL) {
+    struct stat s;
+    if (stat(configDir, &s) != 0 || (!S_ISREG(s.st_mode) && !S_ISLNK(s.st_mode))) {
+      //return true to follow behavior before file support
+      return true;
+    }
+    fp = fopen(configDir, "r");
+    if (fp == NULL) {
+      return false;
+    }
+  }
   
   size_t len = 1024;
   line = calloc(1, len);
   
-  if (fp != NULL) {
-    while (!feof(fp)) {
-      memset(line, 0, len);
-
-      option = value = NULL;
-      olen = vlen = 0;
+  while (!feof(fp)) {
+    memset(line, 0, len);
 
-      getline(&line, &len, fp);
-      line[len - 1] = 0;
-      
-      paGetToken(line, &option, &olen);
-      if (olen == 0) continue;
-      option[olen] = 0;
+    option = value = NULL;
+    olen = vlen = 0;
 
-      paGetToken(option + olen + 1, &value, &vlen);
-      if (vlen == 0) continue;
-      value[vlen] = 0;
+    getline(&line, &len, fp);
+    line[len - 1] = 0;
+    
+    paGetToken(line, &option, &olen);
+    if (olen == 0) continue;
+    option[olen] = 0;
 
-      // For dataDir, the format is:
-      // dataDir    /mnt/disk1    0
-      paGetToken(value + vlen + 1, &value1, &vlen1);
-      
-      taosReadConfigOption(option, value);
-    }
+    paGetToken(option + olen + 1, &value, &vlen);
+    if (vlen == 0) continue;
+    value[vlen] = 0;
 
-    fclose(fp);
+    // For dataDir, the format is:
+    // dataDir    /mnt/disk1    0
+    paGetToken(value + vlen + 1, &value1, &vlen1);
+    
+    taosReadConfigOption(option, value);
   }
 
+  fclose(fp);
+
   tfree(line);
   
   return true;

src/util/src/tkvstore.c

@@ -34,6 +34,7 @@
 #define TD_KVSTORE_MAINOR_VERSION 0
 #define TD_KVSTORE_SNAP_SUFFIX ".snap"
 #define TD_KVSTORE_NEW_SUFFIX ".new"
+#define TD_KVSTORE_INIT_MAGIC 0xFFFFFFFF
 
 typedef struct {
   uint64_t uid;
@@ -140,6 +141,7 @@ SKVStore *tdOpenKVStore(char *fname, iterFunc iFunc, afterFunc aFunc, void *appH
   if (tdLoadKVStoreHeader(pStore->fd, pStore->fname, &info) < 0) goto _err;
 
   pStore->info.size = TD_KVSTORE_HEADER_SIZE;
+  pStore->info.magic = info.magic;
 
   if (tdRestoreKVStore(pStore) < 0) goto _err;
 
@@ -251,6 +253,8 @@ int tdUpdateKVStoreRecord(SKVStore *pStore, uint64_t uid, void *cont, int contLe
     return -1;
   }
 
+  pStore->info.magic =
+      taosCalcChecksum(pStore->info.magic, (uint8_t *)POINTER_SHIFT(cont, contLen - sizeof(TSCKSUM)), sizeof(TSCKSUM));
   pStore->info.size += (sizeof(SKVRecord) + contLen);
   SKVRecord *pRecord = taosHashGet(pStore->map, (void *)&uid, sizeof(uid));
   if (pRecord != NULL) {  // just to insert
@@ -288,6 +292,7 @@ int tdDropKVStoreRecord(SKVStore *pStore, uint64_t uid) {
     return -1;
   }
 
+  pStore->info.magic = taosCalcChecksum(pStore->info.magic, (uint8_t *)buf, POINTER_DISTANCE(pBuf, buf));
   pStore->info.size += POINTER_DISTANCE(pBuf, buf);
   pStore->info.nDels++;
   pStore->info.nRecords--;
@@ -371,7 +376,7 @@ static int tdUpdateKVStoreHeader(int fd, char *fname, SStoreInfo *pInfo) {
 }
 
 static int tdInitKVStoreHeader(int fd, char *fname) {
-  SStoreInfo info = {TD_KVSTORE_HEADER_SIZE, 0, 0, 0};
+  SStoreInfo info = {TD_KVSTORE_HEADER_SIZE, 0, 0, 0, TD_KVSTORE_INIT_MAGIC};
 
   return tdUpdateKVStoreHeader(fd, fname, &info);
 }
@@ -382,6 +387,7 @@ static int tdEncodeStoreInfo(void **buf, SStoreInfo *pInfo) {
   tlen += taosEncodeVariantI64(buf, pInfo->tombSize);
   tlen += taosEncodeVariantI64(buf, pInfo->nRecords);
   tlen += taosEncodeVariantI64(buf, pInfo->nDels);
+  tlen += taosEncodeFixedU32(buf, pInfo->magic);
 
   return tlen;
 }
@@ -391,6 +397,7 @@ static void *tdDecodeStoreInfo(void *buf, SStoreInfo *pInfo) {
   buf = taosDecodeVariantI64(buf, &(pInfo->tombSize));
   buf = taosDecodeVariantI64(buf, &(pInfo->nRecords));
   buf = taosDecodeVariantI64(buf, &(pInfo->nDels));
+  buf = taosDecodeFixedU32(buf, &(pInfo->magic));
 
   return buf;
 }

src/util/src/tutil.c

@@ -522,7 +522,7 @@ int32_t taosFileRename(char *fullPath, char *suffix, char delimiter, char **dstP
 void getTmpfilePath(const char *fileNamePrefix, char *dstPath) {
   const char* tdengineTmpFileNamePrefix = "tdengine-";
   
-  char tmpPath[PATH_MAX] = {0};
+  char tmpPath[PATH_MAX];
 
 #ifdef WINDOWS
   char *tmpDir = getenv("tmp");

src/util/tests/CMakeLists.txt

@@ -11,5 +11,5 @@ IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
     AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 
     ADD_EXECUTABLE(utilTest ${SOURCE_LIST})
-    TARGET_LINK_LIBRARIES(utilTest tutil common gtest pthread)
-ENDIF()
+    TARGET_LINK_LIBRARIES(utilTest tutil common gtest pthread gcov)
+ENDIF()
\ No newline at end of file

tests/examples/c/makefile

@@ -4,7 +4,6 @@
 ROOT=./
 TARGET=exe
 LFLAGS = '-Wl,-rpath,/usr/local/taos/driver/' -ltaos -lpthread -lm -lrt
-#LFLAGS = '-Wl,-rpath,/home/zbm/project/td/debug/build/lib/' -L/home/zbm/project/td/debug/build/lib -ltaos -lpthread -lm -lrt
 CFLAGS = -O3 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion -Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX -msse4.2 -Wno-unused-function -D_M_X64 \
 	-I/usr/local/taos/include -std=gnu99
 
@@ -16,6 +15,7 @@ exe:
 	gcc $(CFLAGS) ./prepare.c -o $(ROOT)/prepare $(LFLAGS)
 	gcc $(CFLAGS) ./stream.c -o $(ROOT)/stream $(LFLAGS)
 	gcc $(CFLAGS) ./subscribe.c -o $(ROOT)subscribe $(LFLAGS)
+	gcc $(CFLAGS) ./apitest.c -o $(ROOT)apitest $(LFLAGS)
 
 clean:
 	rm $(ROOT)/asyncdemo

tests/examples/python/read_example.py

@@ -29,7 +29,7 @@ if __name__ == '__main__':
 
     # Create a database named db
     try:
-        c1.execute('create database db')
+        c1.execute('create database if not exists db ')
     except Exception as err:
         conn.close()
         raise(err)

tests/pytest/fulltest.sh

@@ -144,6 +144,7 @@ python3 ./test.py -f query/querySort.py
 python3 ./test.py -f query/queryJoin.py
 python3 ./test.py -f query/select_last_crash.py
 python3 ./test.py -f query/queryNullValueTest.py
+python3 ./test.py -f query/queryInsertValue.py
 
 #stream
 python3 ./test.py -f stream/metric_1.py
@@ -161,3 +162,22 @@ python3 ./test.py -f client/client.py
 # Misc
 python3 testCompress.py
 python3 testNoCompress.py
+
+# functions
+python3 ./test.py -f functions/function_avg.py
+python3 ./test.py -f functions/function_bottom.py
+python3 ./test.py -f functions/function_count.py
+python3 ./test.py -f functions/function_diff.py
+python3 ./test.py -f functions/function_first.py
+python3 ./test.py -f functions/function_last.py
+python3 ./test.py -f functions/function_last_row.py
+python3 ./test.py -f functions/function_leastsquares.py
+python3 ./test.py -f functions/function_max.py
+python3 ./test.py -f functions/function_min.py
+python3 ./test.py -f functions/function_operations.py
+python3 ./test.py -f functions/function_percentile.py
+python3 ./test.py -f functions/function_spread.py
+python3 ./test.py -f functions/function_stddev.py
+python3 ./test.py -f functions/function_sum.py
+python3 ./test.py -f functions/function_top.py
+python3 ./test.py -f functions/function_twa.py
\ No newline at end of file

tests/pytest/functions/function_avg.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+
+        self.rowNum = 10
+        self.ts = 1537146000000
+        
+    def run(self):
+        tdSql.prepare()
+
+        intData = []        
+        floatData = []
+
+        tdSql.execute('''create table test(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, 
+                    col7 bool, col8 binary(20), col9 nchar(20)) tags(loc nchar(20))''')
+        tdSql.execute("create table test1 using test tags('beijing')")
+        for i in range(self.rowNum):
+            tdSql.execute("insert into test1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')" 
+                        % (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1))
+            intData.append(i + 1)            
+            floatData.append(i + 0.1)                        
+
+        # average verifacation 
+        tdSql.error("select avg(ts) from test")
+        tdSql.error("select avg(ts) from test1")
+        tdSql.error("select avg(col7) from test")
+        tdSql.error("select avg(col7) from test1")
+        tdSql.error("select avg(col8) from test")
+        tdSql.error("select avg(col8) from test1")
+        tdSql.error("select avg(col9) from test")
+        tdSql.error("select avg(col9) from test1")
+
+        tdSql.query("select avg(col1) from test")
+        tdSql.checkData(0, 0, np.average(intData))
+        tdSql.query("select avg(col2) from test")
+        tdSql.checkData(0, 0, np.average(intData))
+        tdSql.query("select avg(col3) from test")
+        tdSql.checkData(0, 0, np.average(intData))
+        tdSql.query("select avg(col4) from test")
+        tdSql.checkData(0, 0, np.average(intData))
+        tdSql.query("select avg(col5) from test")
+        tdSql.checkData(0, 0, np.average(floatData))              
+        tdSql.query("select avg(col6) from test")
+        tdSql.checkData(0, 0, np.average(floatData))              
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/functions/function_bottom.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+
+        self.rowNum = 10
+        self.ts = 1537146000000
+        
+    def run(self):
+        tdSql.prepare()    
+
+        tdSql.execute('''create table test(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, 
+                    col7 bool, col8 binary(20), col9 nchar(20)) tags(loc nchar(20))''')
+        tdSql.execute("create table test1 using test tags('beijing')")
+        for i in range(self.rowNum):
+            tdSql.execute("insert into test1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')" 
+                        % (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1))                            
+
+        # bottom verifacation 
+        tdSql.error("select bottom(ts, 10) from test")
+        tdSql.error("select bottom(col1, 0) from test")
+        tdSql.error("select bottom(col1, 101) from test")
+        tdSql.error("select bottom(col2, 0) from test")
+        tdSql.error("select bottom(col2, 101) from test")
+        tdSql.error("select bottom(col3, 0) from test")
+        tdSql.error("select bottom(col3, 101) from test")
+        tdSql.error("select bottom(col4, 0) from test")
+        tdSql.error("select bottom(col4, 101) from test")
+        tdSql.error("select bottom(col5, 0) from test")
+        tdSql.error("select bottom(col5, 101) from test")
+        tdSql.error("select bottom(col6, 0) from test")
+        tdSql.error("select bottom(col6, 101) from test")        
+        tdSql.error("select bottom(col7, 10) from test")        
+        tdSql.error("select bottom(col8, 10) from test")
+        tdSql.error("select bottom(col9, 10) from test")
+
+        tdSql.query("select bottom(col1, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 2)
+
+        tdSql.query("select bottom(col2, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 2)
+
+        tdSql.query("select bottom(col3, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 2)
+
+        tdSql.query("select bottom(col4, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 1)
+        tdSql.checkData(1, 1, 2)
+
+        tdSql.query("select bottom(col5, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 0.1)
+        tdSql.checkData(1, 1, 1.1)
+
+        tdSql.query("select bottom(col6, 2) from test")
+        tdSql.checkRows(2)
+        tdSql.checkData(0, 1, 0.1)
+        tdSql.checkData(1, 1, 1.1)
+                   
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/functions/function_count.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+
+        self.rowNum = 10
+        self.ts = 1537146000000
+        
+    def run(self):
+        tdSql.prepare()
+
+        tdSql.execute('''create table test(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, 
+                    col7 bool, col8 binary(20), col9 nchar(20)) tags(loc nchar(20))''')
+        tdSql.execute("create table test1 using test tags('beijing')")
+        for i in range(self.rowNum):
+            tdSql.execute("insert into test1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')" 
+                        % (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1))
+        
+        # Count verifacation
+        tdSql.query("select count(*) from test")        
+        tdSql.checkData(0, 0, 10)
+
+        tdSql.query("select count(ts) from test")        
+        tdSql.checkData(0, 0, 10)                        
+        tdSql.query("select count(col1) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col2) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col3) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col4) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col5) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col6) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col7) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col8) from test")        
+        tdSql.checkData(0, 0, 10)
+        tdSql.query("select count(col9) from test")        
+        tdSql.checkData(0, 0, 10)
+
+        tdSql.execute("alter table test add column col10 int")
+        tdSql.query("select count(col10) from test")        
+        tdSql.checkRows(0)
+
+        tdSql.execute("insert into test1 values(now, 1, 2, 3, 4, 1.1, 2.2, false, 'test', 'test' 1)")
+        tdSql.query("select count(col10) from test")        
+        tdSql.checkData(0, 0, 1)        
+              
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/functions/function_diff.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+
+        self.rowNum = 10
+        self.ts = 1537146000000
+        
+    def run(self):
+        tdSql.prepare()
+
+        tdSql.execute('''create table test(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, 
+                    col7 bool, col8 binary(20), col9 nchar(20)) tags(loc nchar(20))''')
+        tdSql.execute("create table test1 using test tags('beijing')")
+        tdSql.execute("insert into test1 values(%d, 0, 0, 0, 0, 0.0, 0.0, False, ' ', ' ')" % (self.ts - 1))
+        
+        # diff verifacation 
+        tdSql.query("select diff(col1) from test1")
+        tdSql.checkRows(0)
+        
+        tdSql.query("select diff(col2) from test1")
+        tdSql.checkRows(0)
+
+        tdSql.query("select diff(col3) from test1")
+        tdSql.checkRows(0)
+
+        tdSql.query("select diff(col4) from test1")
+        tdSql.checkRows(0)
+
+        tdSql.query("select diff(col5) from test1")
+        tdSql.checkRows(0)
+
+        tdSql.query("select diff(col6) from test1")
+        tdSql.checkRows(0)
+
+        for i in range(self.rowNum):
+            tdSql.execute("insert into test1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')" 
+                        % (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1))                                
+        
+        tdSql.error("select diff(ts) from test")
+        tdSql.error("select diff(ts) from test1")
+        tdSql.error("select diff(col1) from test")
+        tdSql.error("select diff(col2) from test")
+        tdSql.error("select diff(col3) from test")
+        tdSql.error("select diff(col4) from test")
+        tdSql.error("select diff(col5) from test")
+        tdSql.error("select diff(col6) from test")
+        tdSql.error("select diff(col7) from test")     
+        tdSql.error("select diff(col7) from test1")               
+        tdSql.error("select diff(col8) from test") 
+        tdSql.error("select diff(col8) from test1")
+        tdSql.error("select diff(col9) from test")        
+        tdSql.error("select diff(col9) from test1")
+        
+        tdSql.query("select diff(col1) from test1")
+        tdSql.checkRows(10)
+
+        tdSql.query("select diff(col2) from test1")
+        tdSql.checkRows(10)
+
+        tdSql.query("select diff(col3) from test1")
+        tdSql.checkRows(10)
+
+        tdSql.query("select diff(col4) from test1")
+        tdSql.checkRows(10)
+
+        tdSql.query("select diff(col5) from test1")
+        tdSql.checkRows(10)
+
+        tdSql.query("select diff(col6) from test1")
+        tdSql.checkRows(10)
+        
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())