Merge branch 'develop' into coverity_scan

.travis.yml

@@ -13,7 +13,7 @@ branches:
 
 matrix:
   - os: linux
-    dist: bionic
+    dist: focal
     language: c
 
     git:
@@ -28,8 +28,6 @@ matrix:
           - build-essential
           - cmake
           - net-tools
-          - python-pip
-          - python-setuptools
           - python3-pip
           - python3-setuptools
           - valgrind
@@ -54,13 +52,19 @@ matrix:
           cd ${TRAVIS_BUILD_DIR}/debug
           make install > /dev/null || travis_terminate $?
 
-          pip install numpy
-          pip install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python2/
           pip3 install numpy
           pip3 install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python3/
 
           cd ${TRAVIS_BUILD_DIR}/tests
           ./test-all.sh smoke || travis_terminate $?
+          sleep 1
+
+          cd ${TRAVIS_BUILD_DIR}/tests/pytest
+          pkill -TERM -x taosd
+          fuser -k -n tcp 6030
+          sleep 1
+          ./crash_gen.sh -a -p -t 4 -s 25|| travis_terminate $?
+          sleep 1
 
           cd ${TRAVIS_BUILD_DIR}/tests/pytest
           ./valgrind-test.sh 2>&1 > mem-error-out.log
@@ -160,7 +164,7 @@ matrix:
 
     script:
       - cmake .. > /dev/null
-      - make > /dev/null
+      - make
 
   - os: linux
     dist: bionic

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

@@ -480,9 +480,9 @@ TDengine支持针对数据的聚合查询。提供支持的聚合和选择函数
 
 - **LEASTSQUARES**
     ```mysql
-    SELECT LEASTSQUARES(field_name) FROM tb_name [WHERE clause]
+    SELECT LEASTSQUARES(field_name, start_val, step_val) FROM tb_name [WHERE clause]
     ```
-    功能说明：统计表中某列的值是主键（时间戳）的拟合直线方程。  
+    功能说明：统计表中某列的值是主键（时间戳）的拟合直线方程。start_val是自变量初始值，step_val是自变量的步长值。  
     返回结果数据类型：字符串表达式（斜率, 截距）。  
     应用字段：不能应用在timestamp、binary、nchar、bool类型字段。  
     说明：自变量是时间戳，因变量是该列的值。  

documentation/webdocs/markdowndocs/TAOS SQL.md

@@ -412,7 +412,7 @@ TDengine supports aggregations over numerical values, they are listed below:
     SELECT PERCENTILE(field_name, P) FROM { tb_name | stb_name } [WHERE clause]
     ```
     Function: the value of the specified column below which `P` percent of the data points fall.  
-    Return Data Type: the same data type.  
+    Return Data Type: double.  
     Applicable Data Types: all types except `timestamp`, `binary`, `nchar`, `bool`. 
     Applied to: table/STable.  
     Note: The range of `P` is `[0, 100]`. When `P=0` , `PERCENTILE` returns the equal value as `MIN`; when `P=100`, `PERCENTILE` returns the equal value as `MAX`. 
@@ -446,7 +446,7 @@ TDengine supports aggregations over numerical values, they are listed below:
     SELECT SPREAD(field_name) FROM { tb_name | stb_name } [WHERE clause]
     ```
     Function: return the difference between the maximum and the mimimum value.  
-    Return Data Type: the same data type.  
+    Return Data Type: double.  
     Applicable Data Types: all types except `timestamp`, `binary`, `nchar`, `bool`.  
     Applied to: table/STable.  
     Note: spread gives the range of data variation in a table/supertable; it is equivalent to `MAX()` - `MIN()`

importSampleData/app/main.go

@@ -18,7 +18,7 @@ import (
 	"sync"
 	"time"
 
-	_ "github.com/taosdata/TDengine/src/connector/go/taosSql"
+	_ "github.com/taosdata/driver-go/taosSql"
 )
 
 const (

src/client/inc/tscLog.h

@@ -31,9 +31,7 @@ extern int32_t tscEmbedded;
 #define tscInfo(...)  { if (cDebugFlag & DEBUG_INFO)  { taosPrintLog("TSC INFO  ", tscEmbedded ? 255 : cDebugFlag, __VA_ARGS__); }}
 #define tscDebug(...) { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC DEBUG ", cDebugFlag, __VA_ARGS__); }}
 #define tscTrace(...) { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC TRACE ", cDebugFlag, __VA_ARGS__); }}
-
-#define tscDebugDump(...) { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC DEBUG ", cDebugFlag, __VA_ARGS__); }}
-#define tscTraceDump(...) { if (cDebugFlag & DEBUG_TRACE) { taosPrintLongString("TSC TRACE ", cDebugFlag, __VA_ARGS__); }}
+#define tscDebugL(...){ if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC DEBUG ", cDebugFlag, __VA_ARGS__); }}
 
 #ifdef __cplusplus
 }

src/client/inc/tscUtil.h

@@ -87,6 +87,16 @@ typedef struct SVgroupTableInfo {
   SArray*       itemList;   //SArray<STableIdInfo>
 } SVgroupTableInfo;
 
+static FORCE_INLINE SQueryInfo* tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex) {
+  assert(pCmd != NULL && subClauseIndex >= 0 && subClauseIndex < TSDB_MAX_UNION_CLAUSE);
+
+  if (pCmd->pQueryInfo == NULL || subClauseIndex >= pCmd->numOfClause) {
+    return NULL;
+  }
+
+  return pCmd->pQueryInfo[subClauseIndex];
+}
+
 int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
                            STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
 void tscDestroyDataBlock(STableDataBlocks* pDataBlock);

src/client/inc/tschemautil.h

@@ -110,8 +110,6 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
 //todo tags value as well as the table id structure needs refactor
 char *tsGetTagsValue(STableMeta *pMeta);
 
-void extractTableNameFromToken(SSQLToken *pToken, SSQLToken* pTable);
-
 #ifdef __cplusplus
 }
 #endif

src/client/inc/tsclient.h

@@ -320,6 +320,8 @@ typedef struct SSqlStream {
   SSqlObj *pSql;
   uint32_t streamId;
   char     listed;
+  bool     isProject;
+  int16_t  precision;
   int64_t  num;  // number of computing count
 
   /*
@@ -334,7 +336,6 @@ typedef struct SSqlStream {
   int64_t etime;     // stream end query time, when time is larger then etime, the stream will be closed
   int64_t interval;
   int64_t slidingTime;
-  int16_t precision;
   void *  pTimer;
 
   void (*fp)();

src/client/src/TSDBJNIConnector.c

@@ -29,9 +29,6 @@
 #define jniDebug(...) { if (jniDebugFlag & DEBUG_DEBUG) { taosPrintLog("JNI DEBUG ", jniDebugFlag, __VA_ARGS__); }}
 #define jniTrace(...) { if (jniDebugFlag & DEBUG_TRACE) { taosPrintLog("JNI TRACE ", jniDebugFlag, __VA_ARGS__); }}
 
-#define jniDebugDump(...) { if (jniDebugFlag & DEBUG_DEBUG) { taosPrintLongString("JNI DEBUG ", jniDebugFlag, __VA_ARGS__); }}
-#define jniTraceDump(...) { if (jniDebugFlag & DEBUG_TRACE) { taosPrintLongString("JNI DEBUG ", jniDebugFlag, __VA_ARGS__); }}
-
 int __init = 0;
 
 JavaVM *g_vm = NULL;

src/client/src/tscAsync.c

@@ -55,7 +55,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
 
   strtolower(pSql->sqlstr, sqlstr);
 
-  tscDebugDump("%p SQL: %s", pSql, pSql->sqlstr);
+  tscDebugL("%p SQL: %s", pSql, pSql->sqlstr);
   pSql->cmd.curSql = pSql->sqlstr;
 
   int32_t code = tsParseSql(pSql, true);
@@ -471,13 +471,19 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
         }
 
         // in case of insert, redo parsing the sql string and build new submit data block for two reasons:
-        // 1. the table Id(tid & uid) may have been update, the submit block needs to be updated
+        // 1. the table Id(tid & uid) may have been update, the submit block needs to be updated accordingly.
         // 2. vnode may need the schema information along with submit block to update its local table schema.
         if (pCmd->command == TSDB_SQL_INSERT) {
           tscDebug("%p redo parse sql string to build submit block", pSql);
 
           pCmd->parseFinished = false;
-          if ((code = tsParseSql(pSql, true)) == TSDB_CODE_SUCCESS) {
+          code = tsParseSql(pSql, true);
+
+          if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+            return;
+          }
+
+          if (code == TSDB_CODE_SUCCESS) {
             /*
              * Discard previous built submit blocks, and then parse the sql string again and build up all submit blocks,
              * and send the required submit block according to index value in supporter to server.

src/client/src/tscFunctionImpl.c

@@ -14,7 +14,6 @@
  */
 
 #include "os.h"
-#include "qast.h"
 #include "qextbuffer.h"
 #include "qfill.h"
 #include "qhistogram.h"
@@ -23,6 +22,7 @@
 #include "qtsbuf.h"
 #include "taosdef.h"
 #include "taosmsg.h"
+#include "qast.h"
 #include "tscLog.h"
 #include "tscSubquery.h"
 #include "tscompression.h"
@@ -340,13 +340,12 @@ bool stableQueryFunctChanged(int32_t funcId) {
  */
 void resetResultInfo(SResultInfo *pResInfo) { pResInfo->initialized = false; }
 
-void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable) {
+void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf) {
   assert(pResInfo->interResultBuf == NULL);
   
   pResInfo->bufLen = size;
   pResInfo->superTableQ = superTable;
-  
-  pResInfo->interResultBuf = calloc(1, (size_t)size);
+  pResInfo->interResultBuf = buf;
 }
 
 // set the query flag to denote that query is completed

src/client/src/tscLocal.c

@@ -406,7 +406,7 @@ int tscProcessLocalCmd(SSqlObj *pSql) {
     pSql->res.qhandle = 0x1;
     pSql->res.numOfRows = 0;
   } else if (pCmd->command == TSDB_SQL_RESET_CACHE) {
-    taosCacheEmpty(tscCacheHandle,false);
+    taosCacheEmpty(tscCacheHandle);
   } else if (pCmd->command == TSDB_SQL_SERV_VERSION) {
     tscProcessServerVer(pSql);
   } else if (pCmd->command == TSDB_SQL_CLI_VERSION) {

src/client/src/tscLocalMerge.c

@@ -274,6 +274,10 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
   pReducer->numOfBuffer = idx;
 
   SCompareParam *param = malloc(sizeof(SCompareParam));
+  if (param == NULL) {
+    tfree(pReducer);
+    return;
+  }
   param->pLocalData = pReducer->pLocalDataSrc;
   param->pDesc = pReducer->pDesc;
   param->num = pReducer->pLocalDataSrc[0]->pMemBuffer->numOfElemsPerPage;
@@ -284,6 +288,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
 
   pRes->code = tLoserTreeCreate(&pReducer->pLoserTree, pReducer->numOfBuffer, param, treeComparator);
   if (pReducer->pLoserTree == NULL || pRes->code != 0) {
+    tfree(param);
     tfree(pReducer);
     return;
   }
@@ -332,6 +337,8 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
     tfree(pReducer->pResultBuf);
     tfree(pReducer->pFinalRes);
     tfree(pReducer->prevRowOfInput);
+    tfree(pReducer->pLoserTree);
+    tfree(param);
     tfree(pReducer);
     pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return;
@@ -364,7 +371,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
   
   TSKEY stime = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
   int64_t revisedSTime =
-      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
+      taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
   
   if (pQueryInfo->fillType != TSDB_FILL_NONE) {
     SFillColInfo* pFillCol = createFillColInfo(pQueryInfo);
@@ -831,7 +838,7 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
   if (pFillInfo != NULL) {
     int64_t stime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.skey : pQueryInfo->window.ekey;
     int64_t revisedSTime =
-        taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
+        taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
   
     taosResetFillInfo(pFillInfo, revisedSTime);
   }
@@ -1301,9 +1308,7 @@ static void resetEnvForNewResultset(SSqlRes *pRes, SSqlCmd *pCmd, SLocalReducer
   if (pQueryInfo->fillType != TSDB_FILL_NONE) {
     TSKEY skey = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
     int64_t newTime =
-        taosGetIntervalStartTimestamp(skey, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
-//    taosResetFillInfo(pLocalReducer->pFillInfo, pQueryInfo->order.order, newTime,
-//                        pQueryInfo->groupbyExpr.numOfGroupCols, 4096, 0, NULL, pLocalReducer->rowSize);
+        taosGetIntervalStartTimestamp(skey, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
     taosResetFillInfo(pLocalReducer->pFillInfo, newTime);
   }
 }

src/client/src/tscParseInsert.c

@@ -1310,6 +1310,11 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
     tscDebug("%p resume to parse sql: %s", pSql, pCmd->curSql);
   }
   
+  ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
+  if (TSDB_CODE_SUCCESS != ret) {
+    return ret;
+  }
+
   if (tscIsInsertData(pSql->sqlstr)) {
     /*
      * Set the fp before parse the sql string, in case of getTableMeta failed, in which
@@ -1326,11 +1331,6 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
     
     ret = tsParseInsertSql(pSql);
   } else {
-    ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
-    if (TSDB_CODE_SUCCESS != ret) {
-      return ret;
-    }
-
     SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
     ret = tscToSQLCmd(pSql, &SQLInfo);
     SQLInfoDestroy(&SQLInfo);

src/client/src/tscPrepare.c

@@ -538,7 +538,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
   pRes->numOfRows = 1;
   
   strtolower(pSql->sqlstr, sql);
-  tscDebugDump("%p SQL: %s", pSql, pSql->sqlstr);
+  tscDebugL("%p SQL: %s", pSql, pSql->sqlstr);
 
   if (tscIsInsertData(pSql->sqlstr)) {  
     pStmt->isInsert = true;

src/client/src/tscSQLParser.c

@@ -18,19 +18,19 @@
 #define _DEFAULT_SOURCE
 
 #include "os.h"
-#include "qast.h"
 #include "taos.h"
 #include "taosmsg.h"
-#include "tstoken.h"
-#include "tstrbuild.h"
-#include "ttime.h"
+#include "qast.h"
+#include "tcompare.h"
+#include "tname.h"
 #include "tscLog.h"
 #include "tscUtil.h"
 #include "tschemautil.h"
 #include "tsclient.h"
+#include "tstoken.h"
+#include "tstrbuild.h"
+#include "ttime.h"
 #include "ttokendef.h"
-#include "tname.h"
-#include "tcompare.h"
 
 #define DEFAULT_PRIMARY_TIMESTAMP_COL_NAME "_c0"
 
@@ -90,6 +90,7 @@ static int32_t validateSqlFunctionInStreamSql(SSqlCmd* pCmd, SQueryInfo* pQueryI
 static int32_t buildArithmeticExprString(tSQLExpr* pExpr, char** exprString);
 static int32_t validateFunctionsInIntervalOrGroupbyQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
 static int32_t validateArithmeticSQLExpr(SSqlCmd* pCmd, tSQLExpr* pExpr, SQueryInfo* pQueryInfo, SColumnList* pList, int32_t* type);
+static int32_t validateEp(char* ep);
 static int32_t validateDNodeConfig(tDCLSQL* pOptions);
 static int32_t validateLocalConfig(tDCLSQL* pOptions);
 static int32_t validateColumnName(char* name);
@@ -359,6 +360,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
 
     case TSDB_SQL_CFG_DNODE: {
       const char* msg2 = "invalid configure options or values";
+      const char* msg3 = "invalid dnode ep";
 
       /* validate the ip address */
       tDCLSQL* pDCL = pInfo->pDCLInfo;
@@ -375,6 +377,10 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       
       strncpy(pCfg->ep, pDCL->a[0].z, pDCL->a[0].n);
 
+      if (validateEp(pCfg->ep) != TSDB_CODE_SUCCESS) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
+      }
+
       strncpy(pCfg->config, pDCL->a[1].z, pDCL->a[1].n);
 
       if (pDCL->nTokens == 3) {
@@ -654,11 +660,14 @@ int32_t parseIntervalClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQ
 int32_t parseSlidingClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
   const char* msg0 = "sliding value too small";
   const char* msg1 = "sliding value no larger than the interval value";
+  const char* msg2 = "sliding value can not less than 1% of interval value";
+
+  const static int32_t INTERVAL_SLIDING_FACTOR = 100;
 
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
-  SSQLToken*      pSliding = &pQuerySql->sliding;
   STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
 
+  SSQLToken* pSliding = &pQuerySql->sliding;
   if (pSliding->n != 0) {
     getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->slidingTime);
     if (tinfo.precision == TSDB_TIME_PRECISION_MILLI) {
@@ -676,6 +685,10 @@ int32_t parseSlidingClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQu
     pQueryInfo->slidingTime = pQueryInfo->intervalTime;
   }
 
+  if ((pQueryInfo->intervalTime != 0) && (pQueryInfo->intervalTime/pQueryInfo->slidingTime > INTERVAL_SLIDING_FACTOR)) {
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -4487,10 +4500,12 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
 
     SUpdateTableTagValMsg* pUpdateMsg = (SUpdateTableTagValMsg*) pCmd->payload;
     pUpdateMsg->head.vgId = htonl(pTableMeta->vgroupInfo.vgId);
-    pUpdateMsg->tid = htonl(pTableMeta->sid);
-    pUpdateMsg->uid = htobe64(pTableMeta->uid);
-    pUpdateMsg->colId = htons(pTagsSchema->colId);
-    pUpdateMsg->tversion = htons(pTableMeta->tversion);
+    pUpdateMsg->tid       = htonl(pTableMeta->sid);
+    pUpdateMsg->uid       = htobe64(pTableMeta->uid);
+    pUpdateMsg->colId     = htons(pTagsSchema->colId);
+    pUpdateMsg->type      = pTagsSchema->type;
+    pUpdateMsg->bytes     = htons(pTagsSchema->bytes);
+    pUpdateMsg->tversion  = htons(pTableMeta->tversion);
     pUpdateMsg->numOfTags = htons(numOfTags);
     pUpdateMsg->schemaLen = htonl(schemaLen);
 
@@ -4627,6 +4642,24 @@ typedef struct SDNodeDynConfOption {
   int32_t len;   // name string length
 } SDNodeDynConfOption;
 
+
+int32_t validateEp(char* ep) {
+  char buf[TSDB_EP_LEN + 1] = {0};
+  tstrncpy(buf, ep, TSDB_EP_LEN);
+
+  char *pos = strchr(buf, ':');
+  if (NULL == pos) {   
+    return TSDB_CODE_TSC_INVALID_SQL;
+  }
+  
+  uint16_t port = atoi(pos+1);
+  if (0 == port) {   
+    return TSDB_CODE_TSC_INVALID_SQL;
+  }  
+
+  return TSDB_CODE_SUCCESS;   
+}
+
 int32_t validateDNodeConfig(tDCLSQL* pOptions) {
   if (pOptions->nTokens < 2 || pOptions->nTokens > 3) {
     return TSDB_CODE_TSC_INVALID_SQL;
@@ -6094,16 +6127,12 @@ int32_t exprTreeFromSqlExpr(SSqlCmd* pCmd, tExprNode **pExpr, const tSQLExpr* pS
       }
     }
 
-
+    // NOTE: binary|nchar data allows the >|< type filter
     if ((*pExpr)->_node.optr != TSDB_RELATION_EQUAL && (*pExpr)->_node.optr != TSDB_RELATION_NOT_EQUAL) {
       if (pRight->nodeType == TSQL_NODE_VALUE) {
         if (pRight->pVal->nType == TSDB_DATA_TYPE_BOOL) {
           return TSDB_CODE_TSC_INVALID_SQL;
         }
-        if ((pRight->pVal->nType == TSDB_DATA_TYPE_BINARY || pRight->pVal->nType == TSDB_DATA_TYPE_NCHAR)
-            && (*pExpr)->_node.optr != TSDB_RELATION_LIKE) {
-          return TSDB_CODE_TSC_INVALID_SQL;
-        }
       }
     }
   }

src/client/src/tscSchemaUtil.c

@@ -215,25 +215,3 @@ __attribute__ ((unused)) static FORCE_INLINE size_t copy(char* dst, const char*
   return len;
 }
 
-/*
- * tablePrefix.columnName
- * extract table name and save it in pTable, with only column name in pToken
- */
-void extractTableNameFromToken(SSQLToken* pToken, SSQLToken* pTable) {
-  const char sep = TS_PATH_DELIMITER[0];
-
-  if (pToken == pTable || pToken == NULL || pTable == NULL) {
-    return;
-  }
-
-  char* r = strnchr(pToken->z, sep, pToken->n, false);
-
-  if (r != NULL) {  // record the table name token
-    pTable->n = r - pToken->z;
-    pTable->z = pToken->z;
-
-    r += 1;
-    pToken->n -= (r - pToken->z);
-    pToken->z = r;
-  }
-}

src/client/src/tscServer.c

@@ -247,7 +247,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
   } else {
     STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
     if (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_RPC_NETWORK_UNAVAIL || rpcMsg->code == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
       if (pCmd->command == TSDB_SQL_CONNECT) {
         rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
         rpcFreeCont(rpcMsg->pCont);
@@ -260,7 +260,12 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
         // get table meta query will not retry, do nothing
       } else {
         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);
@@ -433,8 +438,9 @@ void tscKillSTableQuery(SSqlObj *pSql) {
      * here, we cannot set the command = TSDB_SQL_KILL_QUERY. Otherwise, it may cause
      * sub-queries not correctly released and master sql object of super table query reaches an abnormal state.
      */
-    pSql->pSubs[i]->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
-    rpcCancelRequest(pSql->pSubs[i]->pRpcCtx);
+    rpcCancelRequest(pSub->pRpcCtx);
+    pSub->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
+    tscQueueAsyncRes(pSub);
   }
 
   /*
@@ -1950,7 +1956,7 @@ int tscProcessUseDbRsp(SSqlObj *pSql) {
 }
 
 int tscProcessDropDbRsp(SSqlObj *UNUSED_PARAM(pSql)) {
-  taosCacheEmpty(tscCacheHandle, false);
+  taosCacheEmpty(tscCacheHandle);
   return 0;
 }
 
@@ -1996,7 +2002,7 @@ int tscProcessAlterTableMsgRsp(SSqlObj *pSql) {
 
     if (isSuperTable) {  // if it is a super table, reset whole query cache
       tscDebug("%p reset query cache since table:%s is stable", pSql, pTableMetaInfo->name);
-      taosCacheEmpty(tscCacheHandle, false);
+      taosCacheEmpty(tscCacheHandle);
     }
   }
 

src/client/src/tscSql.c

@@ -617,19 +617,18 @@ void taos_stop_query(TAOS_RES *res) {
   if (pSql->signature != pSql) return;
   tscDebug("%p start to cancel query", res);
 
-  pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
 
   SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
   if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
     tscKillSTableQuery(pSql);
-    return;
   }
 
-  if (pSql->cmd.command >= TSDB_SQL_LOCAL) {
-    return;
+  if (pSql->cmd.command < TSDB_SQL_LOCAL) {
+    rpcCancelRequest(pSql->pRpcCtx);
   }
+  pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
+  tscQueueAsyncRes(pSql);
 
-  rpcCancelRequest(pSql->pRpcCtx);
   tscDebug("%p query is cancelled", res);
 }
 

src/client/src/tscStream.c

@@ -71,6 +71,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
 
   pSql->fp = tscProcessStreamQueryCallback;
   pSql->param = pStream;
+  pSql->res.completed = false;
   
   SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@@ -86,7 +87,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
   // failed to get meter/metric meta, retry in 10sec.
   if (code != TSDB_CODE_SUCCESS) {
     int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
-    tscError("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
+    tscDebug("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
     tscSetRetryTimer(pStream, pSql, retryDelayTime);
 
   } else {
@@ -108,7 +109,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
   SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
   tscDebug("%p add into timer", pSql);
 
-  if (isProjectStream(pQueryInfo)) {
+  if (pStream->isProject) {
     /*
      * pQueryInfo->window.ekey, which is the start time, does not change in case of
      * repeat first execution, once the first execution failed.
@@ -121,7 +122,19 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
     }
   } else {
     pQueryInfo->window.skey = pStream->stime - pStream->interval;
-    pQueryInfo->window.ekey = pStream->stime - 1;
+    int64_t etime = taosGetTimestamp(pStream->precision);
+    // delay to wait all data in last time window
+    if (pStream->precision == TSDB_TIME_PRECISION_MICRO) {
+      etime -= tsMaxStreamComputDelay * 1000l;
+    } else {
+      etime -= tsMaxStreamComputDelay;
+    }
+    if (etime > pStream->etime) {
+      etime = pStream->etime;
+    } else {
+      etime = pStream->stime + (etime - pStream->stime) / pStream->interval * pStream->interval;
+    }
+    pQueryInfo->window.ekey = etime;
   }
 
   // launch stream computing in a new thread
@@ -137,7 +150,7 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
   SSqlStream *pStream = (SSqlStream *)param;
   if (tres == NULL || numOfRows < 0) {
     int64_t retryDelay = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
-    tscError("%p stream:%p, query data failed, code:%d, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
+    tscError("%p stream:%p, query data failed, code:0x%08x, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
              retryDelay);
 
     STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pStream->pSql->cmd, 0, 0);
@@ -151,17 +164,45 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
   taos_fetch_rows_a(tres, tscProcessStreamRetrieveResult, param);
 }
 
-static void tscSetTimestampForRes(SSqlStream *pStream, SSqlObj *pSql) {
-  SSqlRes *pRes = &pSql->res;
+// no need to be called as this is alreay done in the query
+static void tscStreamFillTimeGap(SSqlStream* pStream, TSKEY ts) {
+#if 0
+  SSqlObj *   pSql = pStream->pSql;
+  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
+  
+  if (pQueryInfo->fillType != TSDB_FILL_SET_VALUE && pQueryInfo->fillType != TSDB_FILL_NULL) {
+    return;
+  }
 
-  int64_t  timestamp = *(int64_t *)pRes->data;
-  int64_t actualTimestamp = pStream->stime - pStream->interval;
+  SSqlRes *pRes = &pSql->res;
+  /* failed to retrieve any result in this retrieve */
+  pSql->res.numOfRows = 1;
+  void *row[TSDB_MAX_COLUMNS] = {0};
+  char  tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
+  void *oldPtr = pSql->res.data;
+  pSql->res.data = tmpRes;
+  int32_t rowNum = 0;
+
+  while (pStream->stime + pStream->slidingTime < ts) {
+    pStream->stime += pStream->slidingTime;
+    *(TSKEY*)row[0] =  pStream->stime;
+    for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
+      int16_t     offset = tscFieldInfoGetOffset(pQueryInfo, i);
+      TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
+      assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
+      row[i] = pSql->res.data + offset;
+    }
+    (*pStream->fp)(pStream->param, pSql, row);
+    ++rowNum;
+  }
 
-  if (timestamp != actualTimestamp) {
-    // reset the timestamp of each agg point by using start time of each interval
-    *((int64_t *)pRes->data) = actualTimestamp;
-    tscWarn("%p stream:%p, timestamp of points is:%" PRId64 ", reset to %" PRId64, pSql, pStream, timestamp, actualTimestamp);
+  if (rowNum > 0) {
+    tscDebug("%p stream:%p %d rows padded", pSql, pStream, rowNum);
   }
+
+  pRes->numOfRows = 0;
+  pRes->data = oldPtr;
+#endif
 }
 
 static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOfRows) {
@@ -170,7 +211,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
 
   if (pSql == NULL || numOfRows < 0) {
     int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
-    tscError("%p stream:%p, retrieve data failed, code:%d, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
+    tscError("%p stream:%p, retrieve data failed, code:0x%08x, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
   
     tscSetRetryTimer(pStream, pStream->pSql, retryDelayTime);
     return;
@@ -180,16 +221,11 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
 
   if (numOfRows > 0) { // when reaching here the first execution of stream computing is successful.
     pStream->numOfRes += numOfRows;
-    SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
-    
     for(int32_t i = 0; i < numOfRows; ++i) {
       TAOS_ROW row = taos_fetch_row(res);
       tscDebug("%p stream:%p fetch result", pSql, pStream);
-      if (isProjectStream(pQueryInfo)) {
-        pStream->stime = *(TSKEY *)row[0];
-      } else {
-        tscSetTimestampForRes(pStream, pSql);
-      }
+      tscStreamFillTimeGap(pStream, *(TSKEY*)row[0]);
+      pStream->stime = *(TSKEY *)row[0];
 
       // user callback function
       (*pStream->fp)(pStream->param, res, row);
@@ -199,55 +235,18 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
     taos_fetch_rows_a(res, tscProcessStreamRetrieveResult, pStream);
   } else {  // numOfRows == 0, all data has been retrieved
     pStream->useconds += pSql->res.useconds;
-
-    SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
-    
     if (pStream->numOfRes == 0) {
-      if (pQueryInfo->fillType == TSDB_FILL_SET_VALUE || pQueryInfo->fillType == TSDB_FILL_NULL) {
-        SSqlRes *pRes = &pSql->res;
-
-        /* failed to retrieve any result in this retrieve */
-        pSql->res.numOfRows = 1;
-        void *row[TSDB_MAX_COLUMNS] = {0};
-        char  tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
-
-        void *oldPtr = pSql->res.data;
-        pSql->res.data = tmpRes;
-  
-        for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
-          int16_t     offset = tscFieldInfoGetOffset(pQueryInfo, i);
-          TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
-
-          assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
-          row[i] = pSql->res.data + offset;
-        }
-
-        tscSetTimestampForRes(pStream, pSql);
-        row[0] = pRes->data;
-
-        //            char result[512] = {0};
-        //            taos_print_row(result, row, pQueryInfo->fieldsInfo.pFields, pQueryInfo->fieldsInfo.numOfOutput);
-        //            tscInfo("%p stream:%p query result: %s", pSql, pStream, result);
-        tscDebug("%p stream:%p fetch result", pSql, pStream);
-
-        // user callback function
-        (*pStream->fp)(pStream->param, res, row);
-
-        pRes->numOfRows = 0;
-        pRes->data = oldPtr;
-      } else if (isProjectStream(pQueryInfo)) {
+      if (pStream->isProject) {
         /* no resuls in the query range, retry */
         // todo set retry dynamic time
         int32_t retry = tsProjectExecInterval;
-        tscError("%p stream:%p, retrieve no data, code:%d, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
+        tscError("%p stream:%p, retrieve no data, code:0x%08x, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
 
         tscSetRetryTimer(pStream, pStream->pSql, retry);
         return;
       }
-    } else {
-      if (isProjectStream(pQueryInfo)) {
-        pStream->stime += 1;
-      }
+    } else if (pStream->isProject) {
+      pStream->stime += 1;
     }
 
     tscDebug("%p stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql, pStream, pTableMetaInfo->name,
@@ -256,16 +255,18 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
     // release the metric/meter meta information reference, so data in cache can be updated
 
     taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pTableMeta), false);
+    tscFreeSqlResult(pSql);
+    tfree(pSql->pSubs);
+    pSql->numOfSubs = 0;
     tfree(pTableMetaInfo->vgroupList);
     tscSetNextLaunchTimer(pStream, pSql);
   }
 }
 
 static void tscSetRetryTimer(SSqlStream *pStream, SSqlObj *pSql, int64_t timer) {
-  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
   int64_t delay = getDelayValueAfterTimewindowClosed(pStream, timer);
   
-  if (isProjectStream(pQueryInfo)) {
+  if (pStream->isProject) {
     int64_t now = taosGetTimestamp(pStream->precision);
     int64_t etime = now > pStream->etime ? pStream->etime : now;
 
@@ -323,8 +324,7 @@ static int64_t getLaunchTimeDelay(const SSqlStream* pStream) {
 static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
   int64_t timer = 0;
   
-  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
-  if (isProjectStream(pQueryInfo)) {
+  if (pStream->isProject) {
     /*
      * for project query, no mater fetch data successfully or not, next launch will issue
      * more than the sliding time window
@@ -342,7 +342,6 @@ static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
       return;
     }
   } else {
-    pStream->stime += pStream->slidingTime;
     if ((pStream->stime - pStream->interval) >= pStream->etime) {
       tscDebug("%p stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql, pStream,
                pStream->stime, pStream->etime);
@@ -409,14 +408,16 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) {
 
   pStream->slidingTime = pQueryInfo->slidingTime;
   
-  pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
-  pQueryInfo->slidingTime = 0;
+  if (pStream->isProject) {
+    pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
+    pQueryInfo->slidingTime = 0;
+  }
 }
 
 static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) {
   SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
   
-  if (isProjectStream(pQueryInfo)) {
+  if (pStream->isProject) {
     // no data in table, flush all data till now to destination meter, 10sec delay
     pStream->interval = tsProjectExecInterval;
     pStream->slidingTime = tsProjectExecInterval;
@@ -489,7 +490,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
 
   SSqlStream *pStream = (SSqlStream *)calloc(1, sizeof(SSqlStream));
   if (pStream == NULL) {
-    tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
+    tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
     tscFreeSqlObj(pSql);
     return NULL;
   }
@@ -503,7 +504,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   }
   strtolower(pSql->sqlstr, sqlstr);
 
-  tscDebugDump("%p SQL: %s", pSql, pSql->sqlstr);
+  tscDebugL("%p SQL: %s", pSql, pSql->sqlstr);
   tsem_init(&pSql->rspSem, 0, 0);
 
   int32_t code = tsParseSql(pSql, true);
@@ -514,7 +515,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   if (pRes->code != TSDB_CODE_SUCCESS) {
     setErrorInfo(pSql, pRes->code, pCmd->payload);
 
-    tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
+    tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
     tscFreeSqlObj(pSql);
     return NULL;
   }
@@ -523,6 +524,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
   STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
   
+  pStream->isProject = isProjectStream(pQueryInfo);
   pStream->fp = fp;
   pStream->callback = callback;
   pStream->param = param;
@@ -565,6 +567,8 @@ void taos_close_stream(TAOS_STREAM *handle) {
     taosTmrStopA(&(pStream->pTimer));
 
     tscDebug("%p stream:%p is closed", pSql, pStream);
+    // notify CQ to release the pStream object
+    pStream->fp(pStream->param, NULL, NULL);
 
     tscFreeSqlObj(pSql);
     pStream->pSql = NULL;

src/client/src/tscSubquery.c

@@ -14,12 +14,12 @@
  */
 #include "os.h"
 
-#include "tscSubquery.h"
+#include "qtsbuf.h"
 #include "qast.h"
 #include "tcompare.h"
-#include "tschemautil.h"
-#include "qtsbuf.h"
 #include "tscLog.h"
+#include "tscSubquery.h"
+#include "tschemautil.h"
 #include "tsclient.h"
 
 typedef struct SInsertSupporter {
@@ -57,10 +57,15 @@ static int64_t doTSBlockIntersect(SSqlObj* pSql, SJoinSupporter* pSupporter1, SJ
   pSubQueryInfo1->tsBuf = output1;
   pSubQueryInfo2->tsBuf = output2;
 
+  // no result generated, return directly
+  if (pSupporter1->pTSBuf == NULL || pSupporter2->pTSBuf == NULL) {
+    tscDebug("%p at least one ts-comp is empty, 0 for secondary query after ts blocks intersecting", pSql);
+    return 0;
+  }
+
   tsBufResetPos(pSupporter1->pTSBuf);
   tsBufResetPos(pSupporter2->pTSBuf);
 
-  // TODO add more details information
   if (!tsBufNextPos(pSupporter1->pTSBuf)) {
     tsBufFlush(output1);
     tsBufFlush(output2);
@@ -210,6 +215,7 @@ static void tscDestroyJoinSupporter(SJoinSupporter* pSupporter) {
     pSupporter->f = NULL;
   }
 
+  tfree(pSupporter->pIdTagList);
   tscTagCondRelease(&pSupporter->tagCond);
   free(pSupporter);
 }
@@ -420,43 +426,6 @@ static void updateQueryTimeRange(SQueryInfo* pQueryInfo, STimeWindow* win) {
   pQueryInfo->window = *win;
 }
 
-static UNUSED_FUNC void tSIntersectionAndLaunchSecQuery(SJoinSupporter* pSupporter, SSqlObj* pSql) {
-  SSqlObj* pParentSql = pSupporter->pObj;
-  SQueryInfo* pParentQueryInfo = tscGetQueryInfoDetail(&pParentSql->cmd, pParentSql->cmd.clauseIndex);
-  
-//  if (tscNonOrderedProjectionQueryOnSTable(pParentQueryInfo, 0)) {
-//    STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
-//    assert(pQueryInfo->numOfTables == 1);
-//
-//    // for projection query, need to try next vnode
-////        int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
-//    int32_t totalVnode = 0;
-//    if ((++pTableMetaInfo->vgroupIndex) < totalVnode) {
-//      tscDebug("%p current vnode:%d exhausted, try next:%d. total vnode:%d. current numOfRes:%d", pSql,
-//               pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVnode, pRes->numOfTotal);
-//
-//      pSql->cmd.command = TSDB_SQL_SELECT;
-//      pSql->fp = tscJoinQueryCallback;
-//      tscProcessSql(pSql);
-//
-//      return;
-//    }
-//  }
-
-    SJoinSupporter* p1 = pParentSql->pSubs[0]->param;
-    SJoinSupporter* p2 = pParentSql->pSubs[1]->param;
-    
-    STimeWindow win = TSWINDOW_INITIALIZER;
-    int64_t num = doTSBlockIntersect(pParentSql, p1, p2, &win);
-    if (num <= 0) {  // no result during ts intersect
-      tscDebug("%p free all sub SqlObj and quit", pParentSql);
-      freeJoinSubqueryObj(pParentSql);
-    } else {
-      updateQueryTimeRange(pParentQueryInfo, &win);
-      tscLaunchRealSubqueries(pParentSql);
-    }
-}
-
 int32_t tscCompareTidTags(const void* p1, const void* p2) {
   const STidTags* t1 = (const STidTags*) varDataVal(p1);
   const STidTags* t2 = (const STidTags*) varDataVal(p2);
@@ -713,9 +682,12 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
   SArray *s1 = NULL, *s2 = NULL;
   getIntersectionOfTableTuple(pQueryInfo, pParentSql, &s1, &s2);
   if (taosArrayGetSize(s1) == 0 || taosArrayGetSize(s2) == 0) {  // no results,return.
-    tscDebug("%p free all sub SqlObj and quit", pParentSql);
+    tscDebug("%p tag intersect does not generated qualified tables for join, free all sub SqlObj and quit", pParentSql);
     freeJoinSubqueryObj(pParentSql);
 
+    // set no result command
+    pParentSql->cmd.command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
+    (*pParentSql->fp)(pParentSql->param, pParentSql, 0);
   } else {
     // proceed to for ts_comp query
     SSqlCmd* pSubCmd1 = &pParentSql->pSubs[0]->cmd;
@@ -846,7 +818,10 @@ static void tsCompRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
   if (num <= 0) {  // no result during ts intersect
     tscDebug("%p no results generated in ts intersection, free all sub SqlObj and quit", pParentSql);
     freeJoinSubqueryObj(pParentSql);
-    
+
+    // set no result command
+    pParentSql->cmd.command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
+    (*pParentSql->fp)(pParentSql->param, pParentSql, 0);
     return;
   }
 
@@ -1472,9 +1447,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
 static void tscFreeSubSqlObj(SRetrieveSupport *trsupport, SSqlObj *pSql) {
   tscDebug("%p start to free subquery result", pSql);
   
-  if (pSql->res.code == TSDB_CODE_SUCCESS) {
-    taos_free_result(pSql);
-  }
+  taos_free_result(pSql);
   
   tfree(trsupport->localBuffer);
   
@@ -1805,6 +1778,7 @@ static SSqlObj *tscCreateSqlObjForSubquery(SSqlObj *pSql, SRetrieveSupport *trsu
     pSql->pSubs[trsupport->subqueryIndex] = pNew;
   }
   
+  printf("------------alloc:%p\n", pNew);
   return pNew;
 }
 

src/client/src/tscUtil.c

@@ -14,21 +14,21 @@
  */
 
 #include "os.h"
-#include "qast.h"
+#include "hash.h"
+#include "tscUtil.h"
 #include "taosmsg.h"
+#include "qast.h"
 #include "tcache.h"
 #include "tkey.h"
 #include "tmd5.h"
-#include "tscProfile.h"
 #include "tscLocalMerge.h"
+#include "tscLog.h"
+#include "tscProfile.h"
 #include "tscSubquery.h"
 #include "tschemautil.h"
 #include "tsclient.h"
 #include "ttimer.h"
 #include "ttokendef.h"
-#include "tscLog.h"
-#include "tscUtil.h"
-#include "hash.h"
 
 static void freeQueryInfoImpl(SQueryInfo* pQueryInfo);
 static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache);
@@ -579,9 +579,9 @@ static int trimDataBlock(void* pDataBlock, STableDataBlocks* pTableDataBlock, bo
     int32_t numOfCols = tscGetNumOfColumns(pTableDataBlock->pTableMeta);
     for(int32_t j = 0; j < numOfCols; ++j) {
       STColumn* pCol = (STColumn*) pDataBlock;
-      pCol->colId = pSchema[j].colId;
+      pCol->colId = htons(pSchema[j].colId);
       pCol->type  = pSchema[j].type;
-      pCol->bytes = pSchema[j].bytes;
+      pCol->bytes = htons(pSchema[j].bytes);
       pCol->offset = 0;
 
       pDataBlock += sizeof(STColumn);
@@ -663,7 +663,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) {
     }
 
     SSubmitBlk* pBlocks = (SSubmitBlk*) pOneTableBlock->pData;
-    int64_t destSize = dataBuf->size + pOneTableBlock->size + pBlocks->numOfRows * expandSize;
+    int64_t destSize = dataBuf->size + pOneTableBlock->size + pBlocks->numOfRows * expandSize + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
 
     if (dataBuf->nAllocSize < destSize) {
       while (dataBuf->nAllocSize < destSize) {
@@ -691,7 +691,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) {
     tscDebug("%p tableId:%s, sid:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pSql, pOneTableBlock->tableId,
         pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
 
-    int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize);
+    int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
 
     pBlocks->tid = htonl(pBlocks->tid);
     pBlocks->uid = htobe64(pBlocks->uid);
@@ -1464,16 +1464,6 @@ STableMetaInfo* tscGetMetaInfo(SQueryInfo* pQueryInfo, int32_t tableIndex) {
   return pQueryInfo->pTableMetaInfo[tableIndex];
 }
 
-SQueryInfo* tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex) {
-  assert(pCmd != NULL && subClauseIndex >= 0 && subClauseIndex < TSDB_MAX_UNION_CLAUSE);
-
-  if (pCmd->pQueryInfo == NULL || subClauseIndex >= pCmd->numOfClause) {
-    return NULL;
-  }
-
-  return pCmd->pQueryInfo[subClauseIndex];
-}
-
 int32_t tscGetQueryInfoDetailSafely(SSqlCmd* pCmd, int32_t subClauseIndex, SQueryInfo** pQueryInfo) {
   int32_t ret = TSDB_CODE_SUCCESS;
 
@@ -1832,7 +1822,6 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
     STableMeta*  pPrevTableMeta = taosCacheTransfer(tscCacheHandle, (void**)&pPrevInfo->pTableMeta);
     
     SVgroupsInfo* pVgroupsInfo = pPrevInfo->vgroupList;
-    pPrevInfo->vgroupList = NULL;
     pFinalInfo = tscAddTableMetaInfo(pNewQueryInfo, name, pPrevTableMeta, pVgroupsInfo, pTableMetaInfo->tagColList);
   }
 
@@ -2097,7 +2086,7 @@ void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()) {
 }
 
 void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t columnIndex) {
-  SFieldSupInfo* pInfo = taosArrayGet(pFieldInfo->pSupportInfo, columnIndex);//tscFieldInfoGetSupp(pFieldInfo, columnIndex);
+  SFieldSupInfo* pInfo = taosArrayGet(pFieldInfo->pSupportInfo, columnIndex);
   assert(pInfo->pSqlExpr != NULL);
 
   int32_t type = pInfo->pSqlExpr->resType;
@@ -2112,7 +2101,7 @@ void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t column
     if (isNull(pData, type)) {
       pRes->tsrow[columnIndex] = NULL;
     } else {
-      pRes->tsrow[columnIndex] = pData + VARSTR_HEADER_SIZE;
+      pRes->tsrow[columnIndex] = ((tstr*)pData)->data;
     }
   
     if (realLen < pInfo->pSqlExpr->resBytes - VARSTR_HEADER_SIZE) { // todo refactor

src/common/inc/tname.h

@@ -3,6 +3,7 @@
 
 #include "os.h"
 #include "taosmsg.h"
+#include "tstoken.h"
 
 typedef struct SDataStatis {
   int16_t colId;
@@ -23,10 +24,14 @@ void extractTableName(const char *tableId, char *name);
 
 char* extractDBName(const char *tableId, char *name);
 
+void extractTableNameFromToken(SSQLToken *pToken, SSQLToken* pTable);
+
 SSchema tGetTableNameColumnSchema();
 
 bool tscValidateTableNameLength(size_t len);
 
 SColumnFilterInfo* tscFilterInfoClone(const SColumnFilterInfo* src, int32_t numOfFilters);
 
+int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, int64_t intervalTime, char timeUnit, int16_t precision);
+
 #endif  // TDENGINE_NAME_H

src/common/inc/tulog.h

@@ -32,9 +32,6 @@ extern int32_t tscEmbedded;
 #define uDebug(...) { if (uDebugFlag & DEBUG_DEBUG) { taosPrintLog("UTL DEBUG ", uDebugFlag, __VA_ARGS__); }}
 #define uTrace(...) { if (uDebugFlag & DEBUG_TRACE) { taosPrintLog("UTL TRACE ", uDebugFlag, __VA_ARGS__); }}
 
-#define uDebugDump(...) { if (uDebugFlag & DEBUG_DEBUG) { taosPrintLongString("UTL DEBUG ", uDebugFlag, __VA_ARGS__); }}
-#define uTraceDump(...) { if (uDebugFlag & DEBUG_TRACE) { taosPrintLongString("UTL TRACE ", uDebugFlag, __VA_ARGS__); }}
-
 #define pError(...) { taosPrintLog("APP ERROR ", 255, __VA_ARGS__); }  
 #define pPrint(...) { taosPrintLog("APP INFO  ", 255, __VA_ARGS__); }  
 

src/common/src/tglobal.c

@@ -129,7 +129,7 @@ int32_t tsMnodeEqualVnodeNum = 4;
 int32_t  tsEnableHttpModule = 1;
 int32_t  tsRestRowLimit = 10240;
 uint16_t tsHttpPort = 6020;  // only tcp, range tcp[6020]
-int32_t  tsHttpCacheSessions = 100;
+int32_t  tsHttpCacheSessions = 1000;
 int32_t  tsHttpSessionExpire = 36000;
 int32_t  tsHttpMaxThreads = 2;
 int32_t  tsHttpEnableCompress = 0;
@@ -1210,7 +1210,7 @@ void taosInitGlobalCfg() {
 }
 
 bool taosCheckGlobalCfg() {
-  if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG) {
+  if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG || debugFlag & DEBUG_DUMP) {
     taosSetAllDebugFlag();
   }
   

src/common/src/tname.c

@@ -75,3 +75,56 @@ SColumnFilterInfo* tscFilterInfoClone(const SColumnFilterInfo* src, int32_t numO
 
   return pFilter;
 }
+
+int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, int64_t intervalTime, char timeUnit, int16_t precision) {
+  if (slidingTime == 0) {
+    return startTime;
+  }
+
+  int64_t start = ((startTime - intervalTime) / slidingTime + 1) * slidingTime;
+  if (!(timeUnit == 'a' || timeUnit == 'm' || timeUnit == 's' || timeUnit == 'h')) {
+    /*
+     * here we revised the start time of day according to the local time zone,
+     * but in case of DST, the start time of one day need to be dynamically decided.
+     */
+    // todo refactor to extract function that is available for Linux/Windows/Mac platform
+#if defined(WINDOWS) && _MSC_VER >= 1900
+    // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
+    int64_t timezone = _timezone;
+    int32_t daylight = _daylight;
+    char**  tzname = _tzname;
+#endif
+
+    int64_t t = (precision == TSDB_TIME_PRECISION_MILLI) ? MILLISECOND_PER_SECOND : MILLISECOND_PER_SECOND * 1000L;
+    start += timezone * t;
+  }
+
+  int64_t end = start + intervalTime - 1;
+  if (end < startTime) {
+    start += slidingTime;
+  }
+  return start;
+}
+
+/*
+ * tablePrefix.columnName
+ * extract table name and save it in pTable, with only column name in pToken
+ */
+void extractTableNameFromToken(SSQLToken* pToken, SSQLToken* pTable) {
+  const char sep = TS_PATH_DELIMITER[0];
+
+  if (pToken == pTable || pToken == NULL || pTable == NULL) {
+    return;
+  }
+
+  char* r = strnchr(pToken->z, sep, pToken->n, false);
+
+  if (r != NULL) {  // record the table name token
+    pTable->n = r - pToken->z;
+    pTable->z = pToken->z;
+
+    r += 1;
+    pToken->n -= (r - pToken->z);
+    pToken->z = r;
+  }
+}

src/connector/jdbc/buildTDengine.sh

-#!/bin/bash
-ulimit -c unlimited
-
-function buildTDengine {    
-	cd /root/TDengine
-
-	git remote update
-	REMOTE_COMMIT=`git rev-parse --short remotes/origin/develop`
-	LOCAL_COMMIT=`git rev-parse --short @`
-
-	echo " LOCAL: $LOCAL_COMMIT"
-	echo "REMOTE: $REMOTE_COMMIT"
-	if [ "$LOCAL_COMMIT" == "$REMOTE_COMMIT" ]; then
-		echo "repo up-to-date"
-	else
-		echo "repo need to pull"        
-		git pull
-
-		LOCAL_COMMIT=`git rev-parse --short @`        
-		cd /root/TDengine/debug        
-		rm -rf /root/TDengine/debug/*        
-		cmake ..
-		make > /dev/null
-		make install
-	fi
-}
-
-function restartTaosd {
-	systemctl stop taosd
-	pkill -KILL -x taosd
-	sleep 10
-	
-    	logDir=`grep 'logDir' /etc/taos/taos.cfg|awk 'END{print $2}'`
-    	dataDir=`grep 'dataDir' /etc/taos/taos.cfg|awk '{print $2}'`    
-    
-	rm -rf $logDir/*
-	rm -rf $dataDir/*
-	
-	taosd 2>&1 > /dev/null &
-	sleep 10
-}
-
-buildTDengine
-restartTaosd

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

+package com.taosdata.jdbc.utils;
+
+import java.io.File;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+public class TDNode {
+    
+    private int index;
+    private int running;
+    private int deployed;
+    private boolean testCluster;    
+    private String path;    
+    private String cfgDir;
+    private String dataDir;
+    private String logDir;
+    private String cfgPath;
+
+    public TDNode(int index) {
+        this.index = index;
+        running = 0;
+        deployed = 0;
+        testCluster = false;            
+    }
+
+    public void setPath(String path) {
+        this.path = path;
+    }
+
+    public void setTestCluster(boolean testCluster) {
+        this.testCluster = testCluster;
+    }
+
+    public void searchTaosd(File dir, ArrayList<String> taosdPath) {        
+        File[] fileList = dir.listFiles();
+        
+        if(fileList == null || fileList.length == 0) {
+            return;
+        }
+        
+        for(File file : fileList) {
+            if(file.isFile()) {
+                if(file.getName().equals("taosd")) {                        
+                    taosdPath.add(file.getAbsolutePath());
+                }
+            } else {
+                searchTaosd(file, taosdPath);
+            } 
+        }        
+    }
+
+    public void start() {
+        String selfPath = System.getProperty("user.dir");
+        String binPath = "";
+        String projDir = selfPath + "/../../../../";
+
+        try {
+            ArrayList<String> taosdPath = new ArrayList<>();
+            
+            File dir = new File(projDir);
+            String realProjDir = dir.getCanonicalPath();
+            dir = new File(realProjDir);
+            System.out.println("project Dir: " + projDir);
+            searchTaosd(dir, taosdPath);
+                        
+            if(taosdPath.size() == 0) {
+                System.out.println("The project path doens't exist");
+                return;
+            } else {
+                for(String p : taosdPath) {
+                    if(!p.contains("packing")) {
+                        binPath = p;
+                    }
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        if(binPath.isEmpty()) {
+            System.out.println("taosd not found");
+            return;
+        } else {
+            System.out.println("taosd found in " + binPath);
+        }
+
+        if(this.deployed == 0) {
+            System.out.println("dnode" + index + "is not deployed");
+            return;
+        }
+
+        String cmd = "nohup " + binPath + " -c " + cfgDir + " > /dev/null 2>&1 & "; 
+        System.out.println("start taosd cmd: " + cmd);
+        
+        try{
+            Runtime.getRuntime().exec(cmd);  
+            TimeUnit.SECONDS.sleep(5);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        
+        this.running = 1;        
+    }
+
+    public void stop() {
+        String toBeKilled = "taosd";
+
+        if (this.running != 0) {
+            String killCmd = "pkill -kill -x " + toBeKilled;
+            String[] killCmds = {"sh", "-c", killCmd};
+            try {                                
+                Runtime.getRuntime().exec(killCmds).waitFor();
+
+                for(int port = 6030; port < 6041; port ++) {
+                    String fuserCmd = "fuser -k -n tcp " + port;
+                    Runtime.getRuntime().exec(fuserCmd).waitFor();
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+            this.running = 0;
+            System.out.println("dnode:" + this.index + " is stopped by pkill");
+        }
+    }
+
+    public void startIP() {
+        try{
+            String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " up";        
+            Runtime.getRuntime().exec(cmd).waitFor();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void stopIP() {
+        try{
+            String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " down";            
+            Runtime.getRuntime().exec(cmd).waitFor();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void setCfgConfig(String option, String value) {
+        try{
+            String cmd = "echo " + option + " " + value + " >> " + this.cfgPath;
+            String[] cmdLine = {"sh", "-c", cmd};
+            Process ps = Runtime.getRuntime().exec(cmdLine);
+            ps.waitFor();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public String getDnodeRootDir() {
+        String dnodeRootDir = this.path + "/sim/psim/dnode" + this.index;
+        return dnodeRootDir;
+    }
+
+    public String getDnodesRootDir() {
+        String dnodesRootDir = this.path + "/sim/psim" + this.index;
+        return dnodesRootDir;
+    }
+
+    public void deploy() {
+        this.logDir = this.path + "/sim/dnode" + this.index + "/log";
+        this.dataDir = this.path + "/sim/dnode" + this.index + "/data";
+        this.cfgDir = this.path + "/sim/dnode" + this.index + "/cfg";
+        this.cfgPath = this.path + "/sim/dnode" + this.index + "/cfg/taos.cfg";
+
+        try {        
+            String cmd = "rm -rf " + this.logDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+            
+            cmd = "rm -rf " + this.cfgDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+
+            cmd = "rm -rf " + this.dataDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+
+            cmd = "mkdir -p " + this.logDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+
+            cmd = "mkdir -p " + this.cfgDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+
+            cmd = "mkdir -p " + this.dataDir;
+            Runtime.getRuntime().exec(cmd).waitFor();
+
+            cmd = "touch " + this.cfgPath;
+            Runtime.getRuntime().exec(cmd).waitFor();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }        
+
+        if(this.testCluster) {
+            startIP();
+            setCfgConfig("masterIp", "192.168.0.1");
+            setCfgConfig("secondIp", "192.168.0.2");
+            setCfgConfig("publicIp", "192.168.0." + this.index);
+            setCfgConfig("internalIp", "192.168.0." + this.index);
+            setCfgConfig("privateIp", "192.168.0." + this.index);
+        }
+        setCfgConfig("dataDir", this.dataDir);
+        setCfgConfig("logDir", this.logDir);
+        setCfgConfig("numOfLogLines", "1000000/00");
+        setCfgConfig("mnodeEqualVnodeNum", "0");
+        setCfgConfig("walLevel", "1");
+        setCfgConfig("statusInterval", "1");
+        setCfgConfig("numOfTotalVnodes", "64");
+        setCfgConfig("numOfMnodes", "3");
+        setCfgConfig("numOfThreadsPerCore", "2.0");
+        setCfgConfig("monitor", "0");
+        setCfgConfig("maxVnodeConnections", "30000");
+        setCfgConfig("maxMgmtConnections", "30000");
+        setCfgConfig("maxMeterConnections", "30000");
+        setCfgConfig("maxShellConns", "30000");
+        setCfgConfig("locale", "en_US.UTF-8");
+        setCfgConfig("charset", "UTF-8");
+        setCfgConfig("asyncLog", "0");
+        setCfgConfig("anyIp", "0");
+        setCfgConfig("dDebugFlag", "135");
+        setCfgConfig("mDebugFlag", "135");
+        setCfgConfig("sdbDebugFlag", "135");
+        setCfgConfig("rpcDebugFlag", "135");
+        setCfgConfig("tmrDebugFlag", "131");
+        setCfgConfig("cDebugFlag", "135");
+        setCfgConfig("httpDebugFlag", "135");
+        setCfgConfig("monitorDebugFlag", "135");
+        setCfgConfig("udebugFlag", "135");
+        setCfgConfig("jnidebugFlag", "135");
+        setCfgConfig("qdebugFlag", "135"); 
+        this.deployed = 1;
+    }
+}
\ No newline at end of file

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

+package com.taosdata.jdbc.utils;
+
+import java.io.File;
+import java.util.*;
+
+public class TDNodes {
+    private ArrayList<TDNode> tdNodes; 
+    private boolean testCluster;     
+
+    public TDNodes () {
+        tdNodes = new ArrayList<>();
+        for(int i = 1; i < 11; i ++) {            
+            tdNodes.add(new TDNode(i));
+        }               
+    }
+
+    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;
+    }
+
+    public void check(int index) {
+        if(index < 1 || index > 10) {
+            System.out.println("index: " + index + " should on a scale of [1, 10]");
+            return;
+        }
+    } 
+
+    public void deploy(int index) {
+        try {     
+            File file = new File(System.getProperty("user.dir") + "/../../../");
+            String projectRealPath = file.getCanonicalPath();
+            check(index);        
+            tdNodes.get(index - 1).setTestCluster(this.testCluster);            
+            tdNodes.get(index - 1).setPath(projectRealPath); 
+            tdNodes.get(index - 1).deploy();
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.out.println("deploy Test Exception");
+        }
+    }
+    
+    public void cfg(int index, String option, String value) {
+        check(index);
+        tdNodes.get(index - 1).setCfgConfig(option, value);
+    }
+    
+    public void start(int index) {
+        check(index);
+        tdNodes.get(index - 1).start();
+    }
+
+    public void stop(int index) {
+        check(index);
+        tdNodes.get(index - 1).stop();
+    }
+    
+    public void startIP(int index) {
+        check(index);
+        tdNodes.get(index - 1).startIP();
+    }
+
+    public void stopIP(int index) {
+        check(index);
+        tdNodes.get(index - 1).stopIP();
+    }
+
+}
\ No newline at end of file

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

 package com.taosdata.jdbc;
 
-import java.io.BufferedReader;
-import java.io.InputStreamReader;
+import java.io.File;
+import com.taosdata.jdbc.utils.TDNodes;
 
+import org.junit.AfterClass;
 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();    
+    
     
     @BeforeClass
     public static void setupEnv() {
         try{
-            String path = System.getProperty("user.dir");
-            String bashPath = path + "/buildTDengine.sh";
+            File file = new File(deployPath + "/../../../");
+            String rootPath = file.getCanonicalPath();
+                       
+            tdNodes.setPath(rootPath);
+            tdNodes.setTestCluster(testCluster);            
 
-            Process ps = Runtime.getRuntime().exec(bashPath);
-            ps.waitFor();
+            tdNodes.deploy(1);
+            tdNodes.start(1);  
 
-            BufferedReader br = new BufferedReader(new InputStreamReader(ps.getInputStream()));
-            while(br.readLine() != null) {
-                System.out.println(br.readLine());
-            }            
         } catch (Exception e) {
             e.printStackTrace();
+            System.out.println("Base Test Exception");
         }
     }
+
+    @AfterClass
+    public static void cleanUpEnv() {
+        tdNodes.stop(1);
+    }
 }
\ No newline at end of file

src/connector/python/linux/python3/taos/cursor.py

 from .cinterface import CTaosInterface
 from .error import *
 from .constants import FieldType
+import threading
 
 # querySeqNum = 0
 
@@ -37,6 +38,7 @@ class TDengineCursor(object):
         self._block_iter = 0
         self._affected_rows = 0
         self._logfile = ""
+        self._threadId = threading.get_ident()
 
         if connection is not None:
             self._connection = connection
@@ -103,6 +105,12 @@ class TDengineCursor(object):
     def execute(self, operation, params=None):
         """Prepare and execute a database operation (query or command).
         """
+        # if threading.get_ident() != self._threadId:
+        #     info ="Cursor execute:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
+        #     raise OperationalError(info)
+            # print(info)
+            # return None
+
         if not operation:
             return None
 
@@ -188,6 +196,11 @@ class TDengineCursor(object):
     def fetchall(self):
         """Fetch all (remaining) rows of a query result, returning them as a sequence of sequences (e.g. a list of tuples). Note that the cursor's arraysize attribute can affect the performance of this operation.
         """
+        # if threading.get_ident() != self._threadId:
+        #     info ="[WARNING] Cursor fetchall:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
+        #     raise OperationalError(info)
+            # print(info)
+            # return None
         if self._result is None or self._fields is None:
             raise OperationalError("Invalid use of fetchall")
 
@@ -232,6 +245,12 @@ class TDengineCursor(object):
     def _handle_result(self):
         """Handle the return result from query.
         """
+        # if threading.get_ident() != self._threadId:
+        #     info = "Cursor handleresult:Thread ID not match,creater:"+str(self._threadId)+" caller:"+str(threading.get_ident())
+        #     raise OperationalError(info)
+            # print(info)
+            # return None
+
         self._description = []
         for ele in self._fields:
             self._description.append(

src/cq/src/cqMain.c

@@ -109,6 +109,8 @@ void cqClose(void *handle) {
   while (pObj) {
     SCqObj *pTemp = pObj;
     pObj = pObj->next;
+    tdFreeSchema(pTemp->pSchema);
+    tfree(pTemp->sqlStr);
     free(pTemp);
   } 
   
@@ -242,6 +244,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
 
 static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   SCqObj     *pObj = (SCqObj *)param;
+  if (tres == NULL && row == NULL) {
+    pObj->pStream = NULL;
+    return;
+  }
   SCqContext *pContext = pObj->pContext;
   STSchema   *pSchema = pObj->pSchema;
   if (pObj->pStream == NULL) return;
@@ -263,8 +269,14 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
     void* val = row[i];
     if (val == NULL) {
       val = getNullValue(c->type);
-    } else if (IS_VAR_DATA_TYPE(c->type)) {
+    } else if (c->type == TSDB_DATA_TYPE_BINARY) {
       val = ((char*)val) - sizeof(VarDataLenT);
+    } else if (c->type == TSDB_DATA_TYPE_NCHAR) {
+      char buf[TSDB_MAX_NCHAR_LEN];
+      size_t len = taos_fetch_lengths(tres)[i];
+      taosMbsToUcs4(val, len, buf, sizeof(buf), &len);
+      memcpy(val + sizeof(VarDataLenT), buf, len);
+      varDataLen(val) = len;
     }
     tdAppendColVal(trow, val, c->type, c->bytes, c->offset);
   }

src/dnode/inc/dnodeMgmt.h

@@ -22,6 +22,8 @@ extern "C" {
 
 int32_t dnodeInitMgmt();
 void    dnodeCleanupMgmt();
+int32_t dnodeInitMgmtTimer();
+void    dnodeCleanupMgmtTimer();
 void    dnodeDispatchToMgmtQueue(SRpcMsg *rpcMsg);
 
 void*   dnodeGetVnode(int32_t vgId);

src/dnode/src/dnodeMWrite.c

@@ -131,8 +131,8 @@ static void dnodeFreeMnodeWriteMsg(SMnodeMsg *pWrite) {
   taosFreeQitem(pWrite);
 }
 
-void dnodeSendRpcMnodeWriteRsp(void *pRaw, int32_t code) {
-  SMnodeMsg *pWrite = pRaw;
+void dnodeSendRpcMnodeWriteRsp(void *pMsg, int32_t code) {
+  SMnodeMsg *pWrite = pMsg;
   if (pWrite == NULL) return;
   if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) return;
   if (code == TSDB_CODE_MND_ACTION_NEED_REPROCESSED) {

src/dnode/src/dnodeMain.c

@@ -57,6 +57,7 @@ static const SDnodeComponent tsDnodeComponents[] = {
   {"server",  dnodeInitServer,     dnodeCleanupServer},
   {"mgmt",    dnodeInitMgmt,       dnodeCleanupMgmt},
   {"modules", dnodeInitModules,    dnodeCleanupModules},
+  {"mgmt-tmr",dnodeInitMgmtTimer,  dnodeCleanupMgmtTimer},
   {"shell",   dnodeInitShell,      dnodeCleanupShell}
 };
 

src/dnode/src/dnodeMgmt.c

@@ -106,6 +106,12 @@ int32_t dnodeInitMgmt() {
     }
   }
 
+  int32_t code = vnodeInitResources();
+  if (code != TSDB_CODE_SUCCESS) {
+    dnodeCleanupMgmt();
+    return -1;
+  }
+
   // create the queue and thread to handle the message 
   tsMgmtQset = taosOpenQset();
   if (tsMgmtQset == NULL) {
@@ -127,7 +133,7 @@ int32_t dnodeInitMgmt() {
   pthread_attr_init(&thAttr);
   pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
 
-  int32_t code = pthread_create(&tsQthread, &thAttr, dnodeProcessMgmtQueue, NULL);
+  code = pthread_create(&tsQthread, &thAttr, dnodeProcessMgmtQueue, NULL);
   pthread_attr_destroy(&thAttr);
   if (code != 0) {
     dError("failed to create thread to process mgmt queue, reason:%s", strerror(errno));
@@ -141,6 +147,12 @@ int32_t dnodeInitMgmt() {
     return -1;
   }
 
+  dInfo("dnode mgmt is initialized");
+ 
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t dnodeInitMgmtTimer() {
   tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
   if (tsDnodeTmr == NULL) {
     dError("failed to init dnode timer");
@@ -149,13 +161,11 @@ int32_t dnodeInitMgmt() {
   }
 
   taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
-  
-  dInfo("dnode mgmt is initialized");
- 
+  dInfo("dnode mgmt timer is initialized");
   return TSDB_CODE_SUCCESS;
 }
 
-void dnodeCleanupMgmt() {
+void dnodeCleanupMgmtTimer() {
   if (tsStatusTimer != NULL) {
     taosTmrStopA(&tsStatusTimer);
     tsStatusTimer = NULL;
@@ -165,7 +175,10 @@ void dnodeCleanupMgmt() {
     taosTmrCleanUp(tsDnodeTmr);
     tsDnodeTmr = NULL;
   }
+}
 
+void dnodeCleanupMgmt() {
+  dnodeCleanupMgmtTimer();
   dnodeCloseVnodes();
 
   if (tsMgmtQset) taosQsetThreadResume(tsMgmtQset);
@@ -282,13 +295,12 @@ static void *dnodeOpenVnode(void *param) {
 }
 
 static int32_t dnodeOpenVnodes() {
-  int32_t *vnodeList = calloc(TSDB_MAX_VNODES, sizeof(int32_t));
+  int32_t vnodeList[TSDB_MAX_VNODES] = {0};
   int32_t numOfVnodes = 0;
   int32_t status = dnodeGetVnodeList(vnodeList, &numOfVnodes);
 
   if (status != TSDB_CODE_SUCCESS) {
     dInfo("get dnode list failed");
-    free(vnodeList);
     return status;
   }
 
@@ -334,7 +346,6 @@ static int32_t dnodeOpenVnodes() {
     free(pThread->vnodeList);
   }
 
-  free(vnodeList);
   free(threads);
   dInfo("there are total vnodes:%d, openned:%d failed:%d", numOfVnodes, openVnodes, failedVnodes);
 
@@ -342,7 +353,7 @@ static int32_t dnodeOpenVnodes() {
 }
 
 void dnodeStartStream() {
-  int32_t vnodeList[TSDB_MAX_VNODES];
+  int32_t vnodeList[TSDB_MAX_VNODES] = {0};
   int32_t numOfVnodes = 0;
   int32_t status = vnodeGetVnodeList(vnodeList, &numOfVnodes);
 
@@ -359,7 +370,7 @@ void dnodeStartStream() {
 }
 
 static void dnodeCloseVnodes() {
-  int32_t vnodeList[TSDB_MAX_VNODES];
+  int32_t vnodeList[TSDB_MAX_VNODES]= {0};
   int32_t numOfVnodes = 0;
   int32_t status;
 
@@ -398,6 +409,7 @@ static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
 
   void *pVnode = vnodeAcquireVnode(pCreate->cfg.vgId);
   if (pVnode != NULL) {
+    dDebug("vgId:%d, already exist, processed as alter msg", pCreate->cfg.vgId);
     int32_t code = vnodeAlter(pVnode, pCreate);
     vnodeRelease(pVnode);
     return code;

src/dnode/src/dnodeSystem.c

@@ -20,6 +20,7 @@
 #include "tglobal.h"
 #include "dnodeInt.h"
 #include "dnodeMain.h"
+#include "tfile.h"
 
 static void signal_handler(int32_t signum, siginfo_t *sigInfo, void *context);
 static sem_t exitSem;
@@ -67,6 +68,18 @@ int32_t main(int32_t argc, char *argv[]) {
         taosSetAllocMode(TAOS_ALLOC_MODE_DETECT_LEAK, NULL, true);
       }
     }
+#endif
+#ifdef TAOS_RANDOM_FILE_FAIL
+    else if (strcmp(argv[i], "--random-file-fail-factor") == 0) {
+      if ( (i+1) < argc ) {
+        int factor = atoi(argv[i+1]);
+        printf("The factor of random failure is %d\n", factor);
+        taosSetRandomFileFailFactor(factor);
+      } else {
+        printf("Please specify a number for random failure factor!");
+        exit(EXIT_FAILURE);
+      }
+    }
 #endif
   }
 

src/dnode/src/dnodeVWrite.c

@@ -210,6 +210,7 @@ static void *dnodeProcessWriteQueue(void *param) {
   int32_t       numOfMsgs;
   int           type;
   void         *pVnode, *item;
+  SRspRet      *pRspRet;
 
   dDebug("write worker:%d is running", pWorker->workerId);
 
@@ -222,9 +223,11 @@ static void *dnodeProcessWriteQueue(void *param) {
 
     for (int32_t i = 0; i < numOfMsgs; ++i) {
       pWrite = NULL;
+      pRspRet = NULL;
       taosGetQitem(pWorker->qall, &type, &item);
       if (type == TAOS_QTYPE_RPC) {
         pWrite = (SWriteMsg *)item;
+        pRspRet = &pWrite->rspRet;
         pHead = (SWalHead *)(pWrite->pCont - sizeof(SWalHead));
         pHead->msgType = pWrite->rpcMsg.msgType;
         pHead->version = 0;
@@ -234,7 +237,7 @@ static void *dnodeProcessWriteQueue(void *param) {
         pHead = (SWalHead *)item;
       }
 
-      int32_t code = vnodeProcessWrite(pVnode, type, pHead, item);
+      int32_t code = vnodeProcessWrite(pVnode, type, pHead, pRspRet);
       if (pWrite) pWrite->rpcMsg.code = code;
     }
 
@@ -247,6 +250,11 @@ static void *dnodeProcessWriteQueue(void *param) {
       if (type == TAOS_QTYPE_RPC) {
         pWrite = (SWriteMsg *)item;
         dnodeSendRpcVnodeWriteRsp(pVnode, item, pWrite->rpcMsg.code); 
+      } else if (type == TAOS_QTYPE_FWD) {
+        pHead = (SWalHead *)item;
+        vnodeConfirmForward(pVnode, pHead->version, 0);
+        taosFreeQitem(item);
+        vnodeRelease(pVnode);
       } else {
         taosFreeQitem(item);
         vnodeRelease(pVnode);

src/inc/query.h

@@ -87,8 +87,8 @@ int32_t qKillQuery(qinfo_t qinfo);
 void* qOpenQueryMgmt(int32_t vgId);
 void  qSetQueryMgmtClosed(void* pExecutor);
 void  qCleanupQueryMgmt(void* pExecutor);
-void** qRegisterQInfo(void* pMgmt, void* qInfo);
-void** qAcquireQInfo(void* pMgmt, void** key);
+void** qRegisterQInfo(void* pMgmt, uint64_t qInfo);
+void** qAcquireQInfo(void* pMgmt, uint64_t key);
 void** qReleaseQInfo(void* pMgmt, void* pQInfo, bool needFree);
 
 #ifdef __cplusplus

src/inc/taoserror.h

@@ -129,9 +129,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_CLUSTER_CFG_INCONSISTENT, 0, 0x0335, "mnode clus
 
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST,       0, 0x0340, "mnode accounts already exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT,             0, 0x0341, "mnode invalid account")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_PARA,        0, 0x0342, "mnode invalid account parameter")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_OPTION,      0, 0x0343, "mnode invalid acct option")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_ACCTS,           0, 0x0344, "mnode too many accounts")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_OPTION,      0, 0x0342, "mnode invalid acct option")
 
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_USER_ALREADY_EXIST,       0, 0x0350, "mnode user already exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_USER,             0, 0x0351, "mnode invalid user")
@@ -145,7 +143,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_ID,         0, 0x0361, "mnode inva
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_NAME,       0, 0x0362, "mnode invalid table name")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_TYPE,       0, 0x0363, "mnode invalid table type")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TAGS,            0, 0x0364, "mnode too many tags")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TABLES,          0, 0x0365, "mnode too many tables")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TIMESERIES,      0, 0x0366, "mnode not enough time series")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_SUPER_TABLE,          0, 0x0367, "mnode no super table")           // operation only available for super table
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_COL_NAME_TOO_LONG,        0, 0x0368, "mnode column name too long")
@@ -161,13 +158,13 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_OPTION,        0, 0x0382, "mnode inva
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB,               0, 0x0383, "mnode invalid database")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_MONITOR_DB_FORBIDDEN,     0, 0x0384, "mnode monitor db forbidden")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DATABASES,       0, 0x0385, "mnode too many databases")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_DROPPING,           0, 0x0386, "mnode db in dropping")
 
 // dnode
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED,        0, 0x0400, "dnode message not processed")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY,            0, 0x0401, "dnode out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS,          0, 0x0402, "dnode no disk write access")
 TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN,          0, 0x0403, "dnode invalid message length")
-TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_FILE_FORMAT,      0, 0x0404, "dnode invalid file format")
 
 // vnode 
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS,       0, 0x0500, "vnode action in progress")
@@ -180,7 +177,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_DISK_PERMISSIONS,      0, 0x0506, "vnode no d
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_SUCH_FILE_OR_DIR,      0, 0x0507, "vnode no such file or directory")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_OUT_OF_MEMORY,            0, 0x0508, "vnode out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_APP_ERROR,                0, 0x0509, "vnode app error")
-TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH,            0, 0x0214, "vnode no write auth")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_STATUS,           0, 0x0510, "vnode not in ready state")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_SYNCED,               0, 0x0511, "vnode not in synced state")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH,            0, 0x0512, "vnode no write auth")
 
 // tsdb
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_TABLE_ID,         0, 0x0600, "tsdb invalid table id")
@@ -200,6 +199,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_ACTION,           0, 0x060D, "tsdb inval
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_CREATE_TB_MSG,    0, 0x060E, "tsdb invalid create table message")
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_TABLE_DATA_IN_MEM,     0, 0x060F, "tsdb no table data in memory skiplist")
 TAOS_DEFINE_ERROR(TSDB_CODE_TDB_FILE_ALREADY_EXISTS,      0, 0x0610, "tsdb file already exists")
+TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TABLE_RECONFIGURE,        0, 0x0611, "tsdb need to reconfigure table")
 
 // query
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_QHANDLE,          0, 0x0700, "query invalid handle")

src/inc/taosmsg.h

@@ -203,8 +203,7 @@ typedef struct SSubmitBlk {
 typedef struct SSubmitMsg {
   SMsgHead   header;
   int32_t    length;
-  int32_t    compressed : 2;
-  int32_t    numOfBlocks : 30;
+  int32_t    numOfBlocks;
   SSubmitBlk blocks[];
 } SSubmitMsg;
 
@@ -285,6 +284,8 @@ typedef struct {
   int32_t   tid;
   int16_t   tversion;
   int16_t   colId;
+  int8_t    type;
+  int16_t   bytes;
   int32_t   tagValLen;
   int16_t   numOfTags;
   int32_t   schemaLen;
@@ -472,7 +473,7 @@ typedef struct {
 
 typedef struct {
   int32_t  code;
-  uint64_t qhandle;
+  uint64_t qhandle; // query handle
 } SQueryTableRsp;
 
 typedef struct {
@@ -485,7 +486,7 @@ typedef struct SRetrieveTableRsp {
   int32_t numOfRows;
   int8_t  completed;  // all results are returned to client
   int16_t precision;
-  int64_t offset;  // updated offset value for multi-vnode projection query
+  int64_t offset;     // updated offset value for multi-vnode projection query
   int64_t useconds;
   char    data[];
 } SRetrieveTableRsp;

src/inc/tsdb.h

@@ -108,12 +108,14 @@ void tsdbClearTableCfg(STableCfg *config);
 
 void* tsdbGetTableTagVal(const void* pTable, int32_t colId, int16_t type, int16_t bytes);
 char* tsdbGetTableName(void *pTable);
-STableId tsdbGetTableId(void *pTable);
+
+#define TSDB_TABLEID(_table) ((STableId*) (_table))
+
 STableCfg *tsdbCreateTableCfgFromMsg(SMDCreateTableMsg *pMsg);
 
 int   tsdbCreateTable(TSDB_REPO_T *repo, STableCfg *pCfg);
 int   tsdbDropTable(TSDB_REPO_T *pRepo, STableId tableId);
-int   tsdbUpdateTagValue(TSDB_REPO_T *repo, SUpdateTableTagValMsg *pMsg);
+int   tsdbUpdateTableTagValue(TSDB_REPO_T *repo, SUpdateTableTagValMsg *pMsg);
 TSKEY tsdbGetTableLastKey(TSDB_REPO_T *repo, uint64_t uid);
 void  tsdbStartStream(TSDB_REPO_T *repo);
 
@@ -233,9 +235,10 @@ bool tsdbNextDataBlock(TsdbQueryHandleT *pQueryHandle);
  * Get current data block information
  *
  * @param pQueryHandle
+ * @param pBlockInfo
  * @return
  */
-SDataBlockInfo tsdbRetrieveDataBlockInfo(TsdbQueryHandleT *pQueryHandle);
+void tsdbRetrieveDataBlockInfo(TsdbQueryHandleT *pQueryHandle, SDataBlockInfo* pBlockInfo);
 
 /**
  *

src/inc/vnode.h

@@ -60,7 +60,10 @@ void*   vnodeGetWal(void *pVnode);
 int32_t vnodeProcessWrite(void *pVnode, int qtype, void *pHead, void *item);
 int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes);
 void    vnodeBuildStatusMsg(void *param);
+void    vnodeConfirmForward(void *param, uint64_t version, int32_t code);
 void    vnodeSetAccess(SDMVgroupAccess *pAccess, int32_t numOfVnodes);
+
+int32_t vnodeInitResources();
 void    vnodeCleanupResources();
 
 int32_t vnodeProcessRead(void *pVnode, SReadMsg *pReadMsg);

src/kit/shell/src/shellImport.c

@@ -206,9 +206,10 @@ static void shellSourceFile(TAOS *con, char *fptr) {
     
     if (code != 0) {
       fprintf(stderr, "DB error: %s: %s (%d)\n", taos_errstr(con), fname, lineNo);
-      /* free local resouce: allocated memory/metric-meta refcnt */
-      taos_free_result(pSql);
     }
+    
+    /* free local resouce: allocated memory/metric-meta refcnt */
+    taos_free_result(pSql);
 
     memset(cmd, 0, MAX_COMMAND_SIZE);
     cmd_len = 0;

src/kit/taosdemo/taosdemo.c

@@ -520,9 +520,8 @@ int main(int argc, char *argv[]) {
     snprintf(command, BUFFER_SIZE, "create table if not exists %s.meters (ts timestamp%s tags (areaid int, loc binary(10))", db_name, cols);
     queryDB(taos, command);
     printf("meters created!\n");
-
-    taos_close(taos);
   }
+  taos_close(taos);
   
   /* Wait for table to create  */
   multiThreadCreateTable(cols, use_metric, threads, ntables, db_name, tb_prefix, ip_addr, port, user, pass);
@@ -792,9 +791,6 @@ void * createTable(void *sarg)
       snprintf(command, BUFFER_SIZE, "create table if not exists %s.%s%d (ts timestamp%s;", winfo->db_name, winfo->tb_prefix, i, winfo->cols);
       queryDB(winfo->taos, command);
     }
-
-    taos_close(winfo->taos);
-
   } else {
     /* Create all the tables; */
     printf("Creating table from %d to %d\n", winfo->start_table_id, winfo->end_table_id);
@@ -812,7 +808,6 @@ void * createTable(void *sarg)
     }
       queryDB(winfo->taos, command);
     }
-    taos_close(winfo->taos);
   }
 
   return NULL;
@@ -857,7 +852,6 @@ void multiThreadCreateTable(char* cols, bool use_metric, int threads, int ntable
 
   for (int i = 0; i < threads; i++) {
     info *t_info = infos + i;
-    taos_close(t_info->taos);
     sem_destroy(&(t_info->mutex_sem));
     sem_destroy(&(t_info->lock_sem));
   }
@@ -875,6 +869,11 @@ void *readTable(void *sarg) {
   int64_t sTime = rinfo->start_time;
   char *tb_prefix = rinfo->tb_prefix;
   FILE *fp = fopen(rinfo->fp, "a");
+  if (NULL == fp) {
+    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
+    return NULL;
+  }
+  
   int num_of_DPT = rinfo->nrecords_per_table;
   int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
   int totalData = num_of_DPT * num_of_tables;
@@ -930,6 +929,11 @@ void *readMetric(void *sarg) {
   TAOS *taos = rinfo->taos;
   char command[BUFFER_SIZE] = "\0";
   FILE *fp = fopen(rinfo->fp, "a");
+  if (NULL == fp) {
+    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
+    return NULL;
+  }
+  
   int num_of_DPT = rinfo->nrecords_per_table;
   int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
   int totalData = num_of_DPT * num_of_tables;

src/kit/taosmigrate/taosmigrate.c

@@ -40,7 +40,7 @@ struct arguments {
 static error_t parse_opt(int key, char *arg, struct argp_state *state) {
   struct arguments *arguments = state->input;
   switch (key) {
-    case 'w':
+    case 'r':
       arguments->dataDir = arg;
       break;
     case 'd':
@@ -51,6 +51,7 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
       break;
     case 'f':
       arguments->fqdn = arg;
+      break;
     case 'g':
       arguments->dnodeGroups = arg;
       break;

src/kit/taosmigrate/taosmigrateMnodeWal.c

@@ -96,6 +96,7 @@ void walModWalFile(char* walfile) {
   if (wfd < 0) {
     printf("wal:%s, failed to open(%s)\n", newWalFile, strerror(errno));
     free(buffer);
+    close(rfd);
     return ;
   }
 
@@ -116,6 +117,11 @@ void walModWalFile(char* walfile) {
       break;
     } 
 
+    if (pHead->len >= 1024000 - sizeof(SWalHead)) {
+      printf("wal:%s, SWalHead.len(%d) overflow, skip the rest of file\n", walfile, pHead->len);
+      break;
+    } 
+
     ret = read(rfd, pHead->cont, pHead->len);
     if ( ret != pHead->len) {
       printf("wal:%s, failed to read body, skip, len:%d ret:%d\n", walfile, pHead->len, ret);

src/kit/taosmigrate/taosmigrateVnodeCfg.c

@@ -99,6 +99,8 @@ static int32_t readVnodeCfg(SVnodeObj *pVnode, char* cfgFile)
     goto PARSE_OVER;
   }
 
+  content[maxLen] = (char)0;
+
   root = cJSON_Parse(content);
   if (root == NULL) {
     printf("failed to json parse %s, invalid json format\n", cfgFile);

src/mnode/inc/mnodeMnode.h

@@ -44,6 +44,7 @@ void    mnodeDecMnodeRef(struct SMnodeObj *pMnode);
 char *  mnodeGetMnodeRoleStr();
 void    mnodeGetMnodeIpSetForPeer(SRpcIpSet *ipSet);
 void    mnodeGetMnodeIpSetForShell(SRpcIpSet *ipSet);
+char*   mnodeGetMnodeMasterEp();
 
 void    mnodeGetMnodeInfos(void *mnodes);
 void    mnodeUpdateMnodeIpSet();

src/mnode/inc/mnodeSdb.h

@@ -53,6 +53,7 @@ typedef struct {
   void *   rowData;
   int32_t  rowSize;
   int32_t  retCode; // for callback in sdb queue
+  int32_t  processedCount; // for sync fwd callback
   int32_t  (*cb)(struct SMnodeMsg *pMsg, int32_t code);
   struct SMnodeMsg *pMsg;
 } SSdbOper;

src/mnode/src/mnodeAcct.c

@@ -128,6 +128,7 @@ int32_t mnodeInitAccts() {
 void mnodeCleanupAccts() {
   acctCleanUp();
   sdbCloseTable(tsAcctSdb);
+  tsAcctSdb = NULL;
 }
 
 void *mnodeGetAcct(char *name) {

src/mnode/src/mnodeDb.c

@@ -179,9 +179,14 @@ void mnodeDecDbRef(SDbObj *pDb) {
 
 SDbObj *mnodeGetDbByTableId(char *tableId) {
   char db[TSDB_TABLE_ID_LEN], *pos;
-
+ 
+  // tableId format should be :  acct.db.table
   pos = strstr(tableId, TS_PATH_DELIMITER);
+  assert(NULL != pos);
+
   pos = strstr(pos + 1, TS_PATH_DELIMITER);
+  assert(NULL != pos);
+
   memset(db, 0, sizeof(db));
   strncpy(db, tableId, pos - tableId);
 
@@ -459,6 +464,7 @@ void mnodeMoveVgroupToHead(SVgObj *pVgroup) {
 
 void mnodeCleanupDbs() {
   sdbCloseTable(tsDbSdb);
+  tsDbSdb = NULL;
 }
 
 static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
@@ -964,6 +970,11 @@ static int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
     mError("db:%s, failed to alter, invalid db", pAlter->db);
     return TSDB_CODE_MND_INVALID_DB;
   }
+  
+  if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
+    mError("db:%s, status:%d, in dropping", pAlter->db, pMsg->pDb->status);
+    return TSDB_CODE_MND_DB_IN_DROPPING;
+  }
 
   return mnodeAlterDb(pMsg->pDb, pAlter, pMsg);
 }

src/mnode/src/mnodeDnode.c

@@ -88,13 +88,13 @@ static int32_t mnodeDnodeActionDelete(SSdbOper *pOper) {
 }
 
 static int32_t mnodeDnodeActionUpdate(SSdbOper *pOper) {
-  SDnodeObj *pDnode = pOper->pObj;
-  SDnodeObj *pSaved = mnodeGetDnode(pDnode->dnodeId);
-  if (pSaved != NULL && pDnode != pSaved) {
-    memcpy(pSaved, pDnode, pOper->rowSize);
-    free(pDnode);
-    mnodeDecDnodeRef(pSaved);
+  SDnodeObj *pNew = pOper->pObj;
+  SDnodeObj *pDnode = mnodeGetDnode(pNew->dnodeId);
+  if (pDnode != NULL && pNew != pDnode) {
+    memcpy(pDnode, pNew, pOper->rowSize);
+    free(pNew);
   }
+  mnodeDecDnodeRef(pDnode);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -176,6 +176,7 @@ int32_t mnodeInitDnodes() {
 
 void mnodeCleanupDnodes() {
   sdbCloseTable(tsDnodeSdb);
+  tsDnodeSdb = NULL;
 }
 
 void *mnodeGetNextDnode(void *pIter, SDnodeObj **pDnode) { 
@@ -334,7 +335,7 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
   if (pStatus->dnodeId == 0) {
     mDebug("dnode:%d %s, first access", pDnode->dnodeId, pDnode->dnodeEp);
   } else {
-    //mDebug("dnode:%d, status received, access times %d", pDnode->dnodeId, pDnode->lastAccess);
+    mTrace("dnode:%d, status received, access times %d", pDnode->dnodeId, pDnode->lastAccess);
   }
  
   int32_t openVnodes = htons(pStatus->openVnodes);
@@ -468,7 +469,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
   }
 
   mnodeDecDnodeRef(pDnode);
-  if (strcmp(pDnode->dnodeEp, dnodeGetMnodeMasterEp()) == 0) {
+  if (strcmp(pDnode->dnodeEp, mnodeGetMnodeMasterEp()) == 0) {
     mError("dnode:%d, can't drop dnode:%s which is master", pDnode->dnodeId, ep);
     return TSDB_CODE_MND_NO_REMOVE_MASTER;
   }

src/mnode/src/mnodeMain.c

@@ -41,7 +41,7 @@ typedef struct {
   void              (*cleanup)();
 } SMnodeComponent;
 
-void *tsMnodeTmr;
+void *tsMnodeTmr = NULL;
 static bool tsMgmtIsRunning = false;
 
 static const SMnodeComponent tsMnodeComponents[] = {
@@ -121,9 +121,9 @@ void mnodeCleanupSystem() {
   dnodeFreeMnodeWqueue();
   dnodeFreeMnodeRqueue();
   dnodeFreeMnodePqueue();
-  mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1);
   mnodeCleanupTimer();
-
+  mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1);
+  
   mInfo("mnode is cleaned up");
 }
 

src/mnode/src/mnodeMnode.c

@@ -165,6 +165,7 @@ int32_t mnodeInitMnodes() {
 
 void mnodeCleanupMnodes() {
   sdbCloseTable(tsMnodeSdb);
+  tsMnodeSdb = NULL;
   mnodeMnodeDestroyLock();
 }
 
@@ -267,6 +268,10 @@ void mnodeGetMnodeIpSetForShell(SRpcIpSet *ipSet) {
   mnodeMnodeUnLock();
 }
 
+char* mnodeGetMnodeMasterEp() {
+  return tsMnodeInfos.nodeInfos[tsMnodeInfos.inUse].nodeEp;
+}
+
 void mnodeGetMnodeInfos(void *mnodeInfos) {
   mnodeMnodeRdLock();
   *(SDMMnodeInfos *)mnodeInfos = tsMnodeInfos;

src/mnode/src/mnodeSdb.c

@@ -72,8 +72,6 @@ typedef struct {
   void *     sync;
   void *     wal;
   SSyncCfg   cfg;
-  sem_t      sem;
-  int32_t    code;
   int32_t    numOfTables;
   SSdbTable *tableList[SDB_TABLE_MAX];
   pthread_mutex_t mutex;
@@ -201,7 +199,7 @@ static void sdbRestoreTables() {
     sdbDebug("table:%s, is restored, numOfRows:%" PRId64, pTable->tableName, pTable->numOfRows);
   }
 
-  sdbInfo("sdb is restored, version:%" PRId64 " totalRows:%d numOfTables:%d", tsSdbObj.version, totalRows, numOfTables);
+  sdbInfo("sdb is restored, ver:%" PRId64 " totalRows:%d numOfTables:%d", tsSdbObj.version, totalRows, numOfTables);
 }
 
 void sdbUpdateMnodeRoles() {
@@ -244,27 +242,36 @@ static void sdbNotifyRole(void *ahandle, int8_t role) {
   sdbUpdateMnodeRoles();
 }
 
+FORCE_INLINE
 static void sdbConfirmForward(void *ahandle, void *param, int32_t code) {
-  tsSdbObj.code = code;
-  sem_post(&tsSdbObj.sem);
-  sdbDebug("forward request confirmed, version:%" PRIu64 ", result:%s", (int64_t)param, tstrerror(code));
-}
+  assert(param);
+  SSdbOper * pOper = param;
+  SMnodeMsg *pMsg = pOper->pMsg;
+  if (code <= 0) pOper->retCode = code;
+
+  int32_t processedCount = atomic_add_fetch_32(&pOper->processedCount, 1);
+  if (processedCount <= 1) {
+    if (pMsg != NULL) {
+      sdbDebug("app:%p:%p, waiting for confirm this operation, count:%d", pMsg->rpcMsg.ahandle, pMsg, processedCount);
+    }
+    return;
+  }
+
+  if (pMsg != NULL) {
+    sdbDebug("app:%p:%p, is confirmed and will do callback func", pMsg->rpcMsg.ahandle, pMsg);
+  }
 
- static int32_t sdbForwardToPeer(SWalHead *pHead) {
-  if (tsSdbObj.sync == NULL) return TSDB_CODE_SUCCESS;
+  if (pOper->cb != NULL) {
+    pOper->retCode = (*pOper->cb)(pMsg, pOper->retCode);
+  }
 
-  int32_t code = syncForwardToPeer(tsSdbObj.sync, pHead, (void*)pHead->version, TAOS_QTYPE_RPC);
-  if (code > 0) {
-    sdbDebug("forward request is sent, version:%" PRIu64 ", code:%d", pHead->version, code);
-    sem_wait(&tsSdbObj.sem);
-    return tsSdbObj.code;
-  } 
-  return code;
+  dnodeSendRpcMnodeWriteRsp(pMsg, pOper->retCode);
+  taosFreeQitem(pOper);
 }
 
 void sdbUpdateSync() {
   SSyncCfg syncCfg = {0};
-  int32_t index = 0;
+  int32_t  index = 0;
 
   SDMMnodeInfos *mnodes = dnodeGetMnodeInfos();
   for (int32_t i = 0; i < mnodes->nodeNum; ++i) {
@@ -298,7 +305,7 @@ void sdbUpdateSync() {
   }
 
   syncCfg.replica = index;
-  syncCfg.quorum = (syncCfg.replica == 1) ? 1:2;
+  syncCfg.quorum = (syncCfg.replica == 1) ? 1 : 2;
 
   bool hasThisDnode = false;
   for (int32_t i = 0; i < syncCfg.replica; ++i) {
@@ -325,10 +332,10 @@ void sdbUpdateSync() {
   syncInfo.getWalInfo = sdbGetWalInfo;
   syncInfo.getFileInfo = sdbGetFileInfo;
   syncInfo.writeToCache = sdbWriteToQueue;
-  syncInfo.confirmForward = sdbConfirmForward; 
+  syncInfo.confirmForward = sdbConfirmForward;
   syncInfo.notifyRole = sdbNotifyRole;
   tsSdbObj.cfg = syncCfg;
-  
+
   if (tsSdbObj.sync) {
     syncReconfig(tsSdbObj.sync, &syncCfg);
   } else {
@@ -339,7 +346,6 @@ void sdbUpdateSync() {
 
 int32_t sdbInit() {
   pthread_mutex_init(&tsSdbObj.mutex, NULL);
-  sem_init(&tsSdbObj.sem, 0, 0);
 
   if (sdbInitWriteWorker() != 0) {
     return -1;
@@ -367,7 +373,7 @@ void sdbCleanUp() {
   tsSdbObj.status = SDB_STATUS_CLOSING;
   
   sdbCleanupWriteWorker();
-  sdbDebug("sdb will be closed, version:%" PRId64, tsSdbObj.version);
+  sdbDebug("sdb will be closed, ver:%" PRId64, tsSdbObj.version);
 
   if (tsSdbObj.sync) {
     syncStop(tsSdbObj.sync);
@@ -379,7 +385,6 @@ void sdbCleanUp() {
     tsSdbObj.wal = NULL;
   }
   
-  sem_destroy(&tsSdbObj.sem);
   pthread_mutex_destroy(&tsSdbObj.mutex);
 }
 
@@ -466,8 +471,8 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
     atomic_add_fetch_32(&pTable->autoIndex, 1);
   }
 
-  sdbDebug("table:%s, insert record:%s to hash, rowSize:%d numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
-           sdbGetKeyStrFromObj(pTable, pOper->pObj), pOper->rowSize, pTable->numOfRows, sdbGetVersion());
+  sdbDebug("table:%s, insert record:%s to hash, rowSize:%d numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
+           sdbGetKeyStrFromObj(pTable, pOper->pObj), pOper->rowSize, pTable->numOfRows, pOper->pMsg);
 
   (*pTable->insertFp)(pOper);
   return TSDB_CODE_SUCCESS;
@@ -485,8 +490,8 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
   taosHashRemove(pTable->iHandle, key, keySize);
   atomic_sub_fetch_32(&pTable->numOfRows, 1);
   
-  sdbDebug("table:%s, delete record:%s from hash, numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
-           sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
+  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;
@@ -496,8 +501,8 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
 }
 
 static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) {
-  sdbDebug("table:%s, update record:%s in hash, numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
-           sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
+  sdbDebug("table:%s, update record:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
+           sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, pOper->pMsg);
 
   (*pTable->updateFp)(pOper);
   return TSDB_CODE_SUCCESS;
@@ -513,24 +518,22 @@ static int sdbWrite(void *param, void *data, int type) {
   assert(pTable != NULL);
 
   pthread_mutex_lock(&tsSdbObj.mutex);
+  
   if (pHead->version == 0) {
-     // assign version
+    // assign version
     tsSdbObj.version++;
     pHead->version = tsSdbObj.version;
   } else {
     // for data from WAL or forward, version may be smaller
     if (pHead->version <= tsSdbObj.version) {
       pthread_mutex_unlock(&tsSdbObj.mutex);
-      if (type == TAOS_QTYPE_FWD && tsSdbObj.sync != NULL) {
-        sdbDebug("forward request is received, version:%" PRIu64 " confirm it", pHead->version);
-        syncConfirmForward(tsSdbObj.sync, pHead->version, TSDB_CODE_SUCCESS);
-      }
+      sdbDebug("table:%s, failed to restore %s record:%s from source(%d), ver:%" PRId64 " too large, sdb ver:%" PRId64,
+               pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), type, pHead->version, tsSdbObj.version);
       return TSDB_CODE_SUCCESS;
     } else if (pHead->version != tsSdbObj.version + 1) {
       pthread_mutex_unlock(&tsSdbObj.mutex);
-      sdbError("table:%s, failed to restore %s record:%s from wal, version:%" PRId64 " too large, sdb version:%" PRId64,
-               pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version,
-               tsSdbObj.version);
+      sdbError("table:%s, failed to restore %s record:%s from source(%d), ver:%" PRId64 " too large, sdb ver:%" PRId64,
+               pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), type, pHead->version, tsSdbObj.version);
       return TSDB_CODE_MND_APP_ERROR;
     } else {
       tsSdbObj.version = pHead->version;
@@ -542,28 +545,36 @@ static int sdbWrite(void *param, void *data, int type) {
     pthread_mutex_unlock(&tsSdbObj.mutex);
     return code;
   }
-  
-  code = sdbForwardToPeer(pHead);
+
   pthread_mutex_unlock(&tsSdbObj.mutex);
 
   // from app, oper is created
   if (pOper != NULL) {
-    sdbTrace("record from app is disposed, table:%s action:%s record:%s version:%" PRIu64 " result:%s",
-             pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version,
-             tstrerror(code));
-    return code;
+    // forward to peers
+    pOper->processedCount = 0;
+    int32_t syncCode = syncForwardToPeer(tsSdbObj.sync, pHead, pOper, TAOS_QTYPE_RPC);
+    if (syncCode <= 0) pOper->processedCount = 1;
+
+    if (syncCode < 0) {
+      sdbError("table:%s, failed to forward request, result:%s action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
+               tstrerror(syncCode), sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
+    } else if (syncCode > 0) {
+      sdbDebug("table:%s, forward request is sent, action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
+               sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
+    } else {
+      sdbTrace("table:%s, no need to send fwd request, action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
+               sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
+    }
+    return syncCode;
   }
 
-  // from wal or forward msg, oper not created, should add into hash
-  if (tsSdbObj.sync != NULL) {
-    sdbTrace("record from wal forward is disposed, table:%s action:%s record:%s version:%" PRIu64 " confirm it",
-             pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
-    syncConfirmForward(tsSdbObj.sync, pHead->version, code);
-  } else {
-    sdbTrace("record from wal restore is disposed, table:%s action:%s record:%s version:%" PRIu64, pTable->tableName,
-             sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
-  }
+  sdbDebug("table:%s, record from wal/fwd is disposed, action:%s record:%s ver:%" PRId64, pTable->tableName,
+           sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
 
+  // even it is WAL/FWD, it shall be called to update version in sync
+  syncForwardToPeer(tsSdbObj.sync, pHead, pOper, TAOS_QTYPE_RPC);
+
+  // from wal or forward msg, oper not created, should add into hash
   if (action == SDB_ACTION_INSERT) {
     SSdbOper oper = {.rowSize = pHead->len, .rowData = pHead->cont, .table = pTable};
     code = (*pTable->decodeFp)(&oper);
@@ -627,7 +638,7 @@ int32_t sdbInsertRow(SSdbOper *pOper) {
   memcpy(pNewOper, pOper, sizeof(SSdbOper));
 
   if (pNewOper->pMsg != NULL) {
-    sdbDebug("app:%p:%p, table:%s record:%p:%s, insert action is add to sdb queue, ", pNewOper->pMsg->rpcMsg.ahandle,
+    sdbDebug("app:%p:%p, table:%s record:%p:%s, insert action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
              pNewOper->pMsg, pTable->tableName, pOper->pObj, sdbGetKeyStrFromObj(pTable, pOper->pObj));
   }
 
@@ -677,7 +688,7 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
   memcpy(pNewOper, pOper, sizeof(SSdbOper));
 
   if (pNewOper->pMsg != NULL) {
-    sdbDebug("app:%p:%p, table:%s record:%p:%s, delete action is add to sdb queue, ", pNewOper->pMsg->rpcMsg.ahandle,
+    sdbDebug("app:%p:%p, table:%s record:%p:%s, delete action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
              pNewOper->pMsg, pTable->tableName, pOper->pObj, sdbGetKeyStrFromObj(pTable, pOper->pObj));
   }
 
@@ -727,7 +738,7 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
   memcpy(pNewOper, pOper, sizeof(SSdbOper));
 
   if (pNewOper->pMsg != NULL) {
-    sdbDebug("app:%p:%p, table:%s record:%p:%s, update action is add to sdb queue, ", pNewOper->pMsg->rpcMsg.ahandle,
+    sdbDebug("app:%p:%p, table:%s record:%p:%s, update action is add to sdb queue", pNewOper->pMsg->rpcMsg.ahandle,
              pNewOper->pMsg, pTable->tableName, pOper->pObj, sdbGetKeyStrFromObj(pTable, pOper->pObj));
   }
 
@@ -943,20 +954,24 @@ static void *sdbWorkerFp(void *param) {
       taosGetQitem(tsSdbWriteQall, &type, &item);
       if (type == TAOS_QTYPE_RPC) {
         pOper = (SSdbOper *)item;
+        pOper->processedCount = 1;
         pHead = (void *)pOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
+        if (pOper->pMsg != NULL) {
+          sdbDebug("app:%p:%p, table:%s record:%p:%s ver:%" PRIu64 ", will be processed in sdb queue",
+                   pOper->pMsg->rpcMsg.ahandle, pOper->pMsg, ((SSdbTable *)pOper->table)->tableName, pOper->pObj,
+                   sdbGetKeyStr(pOper->table, pHead->cont), pHead->version);
+        }
       } else {
         pHead = (SWalHead *)item;
         pOper = NULL;
       }
 
-      if (pOper != NULL && pOper->pMsg != NULL) {
-        sdbDebug("app:%p:%p, table:%s record:%p:%s version:%" PRIu64 ", will be processed in sdb queue",
-                 pOper->pMsg->rpcMsg.ahandle, pOper->pMsg, ((SSdbTable *)pOper->table)->tableName, pOper->pObj,
-                 sdbGetKeyStr(pOper->table, pHead->cont), pHead->version);
-      }
-
       int32_t code = sdbWrite(pOper, pHead, type);
-      if (pOper) pOper->retCode = code;
+      if (code > 0) code = 0;
+      if (pOper)
+        pOper->retCode = code;
+      else
+        pHead->len = code;  // hackway
     }
 
     walFsync(tsSdbObj.wal);
@@ -965,25 +980,18 @@ static void *sdbWorkerFp(void *param) {
     taosResetQitems(tsSdbWriteQall);
     for (int32_t i = 0; i < numOfMsgs; ++i) {
       taosGetQitem(tsSdbWriteQall, &type, &item);
+
       if (type == TAOS_QTYPE_RPC) {
         pOper = (SSdbOper *)item;
-        if (pOper != NULL && pOper->cb != NULL) {
-          sdbTrace("app:%p:%p, will do callback func, index:%d", pOper->pMsg->rpcMsg.ahandle, pOper->pMsg, i);
-          pOper->retCode = (*pOper->cb)(pOper->pMsg, pOper->retCode);
-        }
-
-        if (pOper != NULL && pOper->pMsg != NULL) {
-          sdbTrace("app:%p:%p, msg is processed, result:%s", pOper->pMsg->rpcMsg.ahandle, pOper->pMsg,
-                   tstrerror(pOper->retCode));
-        }
-
-        if (pOper != NULL) {
-          sdbDecRef(pOper->table, pOper->pObj);
-        }
-
-        dnodeSendRpcMnodeWriteRsp(pOper->pMsg, pOper->retCode);
+        sdbDecRef(pOper->table, pOper->pObj);
+        sdbConfirmForward(NULL, pOper, pOper->retCode);
+      } else if (type == TAOS_QTYPE_FWD) {
+        pHead = (SWalHead *)item;
+        syncConfirmForward(tsSdbObj.sync, pHead->version, pHead->len);
+        taosFreeQitem(item);
+      } else {
+        taosFreeQitem(item);
       }
-      taosFreeQitem(item);
     }
   }
 

src/mnode/src/mnodeShow.c

@@ -307,6 +307,12 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
       code = TSDB_CODE_MND_INVALID_DB;
       goto connect_over;
     }
+    
+    if (pDb->status != TSDB_DB_STATUS_READY) {
+      mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
+      code = TSDB_CODE_MND_DB_IN_DROPPING;
+      goto connect_over;
+    }
     mnodeDecDbRef(pDb);
   }
 
@@ -350,7 +356,12 @@ static int32_t mnodeProcessUseMsg(SMnodeMsg *pMsg) {
   int32_t code = TSDB_CODE_SUCCESS;
   if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pUseDbMsg->db);
   if (pMsg->pDb == NULL) {
-    code = TSDB_CODE_MND_INVALID_DB;
+    return TSDB_CODE_MND_INVALID_DB;
+  }
+  
+  if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
+    mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
+    return TSDB_CODE_MND_DB_IN_DROPPING;
   }
 
   return code;
@@ -403,4 +414,4 @@ void mnodeVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capa
       memmove(data + pShow->offset[i] * rows, data + pShow->offset[i] * capacity, pShow->bytes[i] * rows);
     }
   }
-}
+}
\ No newline at end of file

src/mnode/src/mnodeUser.c

@@ -154,6 +154,7 @@ int32_t mnodeInitUsers() {
 
 void mnodeCleanupUsers() {
   sdbCloseTable(tsUserSdb);
+  tsUserSdb = NULL;
 }
 
 SUserObj *mnodeGetUser(char *name) {

src/plugins/http/inc/httpInt.h

@@ -206,7 +206,7 @@ typedef struct HttpThread {
   pthread_mutex_t threadMutex;
   bool            stop;
   int             pollFd;
-  int             numOfFds;
+  int             numOfContexts;
   int             threadId;
   char            label[HTTP_LABEL_SIZE];
   bool (*processData)(HttpContext *pContext);

src/plugins/http/inc/httpLog.h

@@ -26,8 +26,6 @@ extern int32_t httpDebugFlag;
 #define httpInfo(...)  { if (httpDebugFlag & DEBUG_INFO)  { taosPrintLog("HTP INFO  ", 255, __VA_ARGS__); }}
 #define httpDebug(...) { if (httpDebugFlag & DEBUG_DEBUG) { taosPrintLog("HTP DEBUG ", httpDebugFlag, __VA_ARGS__); }}
 #define httpTrace(...) { if (httpDebugFlag & DEBUG_TRACE) { taosPrintLog("HTP TRACE ", httpDebugFlag, __VA_ARGS__); }}
-
-#define httpDebugDump(...) { if (httpDebugFlag & DEBUG_DEBUG) { taosPrintLongString("HTP DEBUG ", httpDebugFlag, __VA_ARGS__); }}
-#define httpTraceDump(...) { if (httpDebugFlag & DEBUG_TRACE) { taosPrintLongString("HTP TRACE ", httpDebugFlag, __VA_ARGS__); }}
+#define httpTraceL(...){ if (httpDebugFlag & DEBUG_TRACE) { taosPrintLongString("HTP TRACE ", httpDebugFlag, __VA_ARGS__); }}
 
 #endif

src/plugins/http/src/httpHandle.c

@@ -313,9 +313,9 @@ bool httpParseRequest(HttpContext* pContext) {
     return true;
   }
 
-  httpTraceDump("context:%p, fd:%d, ip:%s, thread:%s, numOfFds:%d, read size:%d, raw data:\n%s", pContext, pContext->fd,
-                pContext->ipstr, pContext->pThread->label, pContext->pThread->numOfFds, pContext->parser.bufsize,
-                pContext->parser.buffer);
+  httpTraceL("context:%p, fd:%d, ip:%s, thread:%s, numOfContexts:%d, read size:%d, raw data:\n%s", pContext,
+             pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->pThread->numOfContexts,
+             pContext->parser.bufsize, pContext->parser.buffer);
 
   if (!httpGetHttpMethod(pContext)) {
     return false;

src/plugins/http/src/httpServer.c

@@ -108,7 +108,7 @@ bool httpReadDataImp(HttpContext *pContext) {
 
 static bool httpDecompressData(HttpContext *pContext) {
   if (pContext->contentEncoding != HTTP_COMPRESS_GZIP) {
-    httpTraceDump("context:%p, fd:%d, ip:%s, content:%s", pContext, pContext->fd, pContext->ipstr, pContext->parser.data.pos);
+    httpTraceL("context:%p, fd:%d, ip:%s, content:%s", pContext, pContext->fd, pContext->ipstr, pContext->parser.data.pos);
     return true;
   }
 
@@ -124,8 +124,8 @@ static bool httpDecompressData(HttpContext *pContext) {
   if (ret == 0) {
     memcpy(pContext->parser.data.pos, decompressBuf, decompressBufLen);
     pContext->parser.data.pos[decompressBufLen] = 0;
-    httpTraceDump("context:%p, fd:%d, ip:%s, rawSize:%d, decompressSize:%d, content:%s",
-              pContext, pContext->fd, pContext->ipstr, pContext->parser.data.len, decompressBufLen,  decompressBuf);
+    httpTraceL("context:%p, fd:%d, ip:%s, rawSize:%d, decompressSize:%d, content:%s", pContext, pContext->fd,
+              pContext->ipstr, pContext->parser.data.len, decompressBufLen, decompressBuf);
     pContext->parser.data.len = decompressBufLen;
   } else {
     httpError("context:%p, fd:%d, ip:%s, failed to decompress data, rawSize:%d, error:%d",
@@ -293,7 +293,7 @@ static void *httpAcceptHttpConnection(void *arg) {
 
     totalFds = 1;
     for (int i = 0; i < pServer->numOfThreads; ++i) {
-      totalFds += pServer->pThreads[i].numOfFds;
+      totalFds += pServer->pThreads[i].numOfContexts;
     }
 
     if (totalFds > tsHttpCacheSessions * 100) {
@@ -332,9 +332,9 @@ static void *httpAcceptHttpConnection(void *arg) {
     }
 
     // notify the data process, add into the FdObj list
-    atomic_add_fetch_32(&pThread->numOfFds, 1);
-    httpDebug("context:%p, fd:%d, ip:%s, thread:%s numOfFds:%d totalFds:%d, accept a new connection", pContext, connFd,
-              pContext->ipstr, pThread->label, pThread->numOfFds, totalFds);
+    atomic_add_fetch_32(&pThread->numOfContexts, 1);
+    httpDebug("context:%p, fd:%d, ip:%s, thread:%s numOfContexts:%d totalFds:%d, accept a new connection", pContext,
+              connFd, pContext->ipstr, pThread->label, pThread->numOfContexts, totalFds);
 
     // pick up next thread for next connection
     threadId++;

src/plugins/monitor/src/monitorMain.c

@@ -35,9 +35,6 @@
 #define monitorDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON DEBUG ", monitorDebugFlag, __VA_ARGS__); }}
 #define monitorTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON TRACE ", monitorDebugFlag, __VA_ARGS__); }}
 
-#define monitorDebugDump(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLongString("MON DEBUG ", monitorDebugFlag, __VA_ARGS__); }}
-#define monitorTraceDump(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLongString("MON TRACE ", monitorDebugFlag, __VA_ARGS__); }}
-
 #define SQL_LENGTH     1024
 #define LOG_LEN_STR    100
 #define IP_LEN_STR     18