merge from develop into feature/vnode

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/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/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,7 +471,7 @@ 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);

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"

src/client/src/tscLocalMerge.c

@@ -364,7 +364,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 +831,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 +1301,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/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) {
@@ -4487,10 +4493,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 +4635,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;

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);
   }
 
   /*

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

@@ -503,7 +503,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);

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;
   }
 

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);

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

@@ -1215,7 +1215,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 - slidingTime) / 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/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/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/dnodeMgmt.c

@@ -402,6 +402,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/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/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")
@@ -167,7 +164,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED,        0, 0x0400, "dnode mess
 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 +176,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 +198,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;

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,6 +60,7 @@ 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();

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;
@@ -875,6 +870,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 +930,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/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

@@ -456,6 +456,7 @@ void mnodeRemoveVgroupFromDb(SVgObj *pVgroup) {
 
 void mnodeCleanupDbs() {
   sdbCloseTable(tsDbSdb);
+  tsDbSdb = NULL;
 }
 
 static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {

src/mnode/src/mnodeDnode.c

@@ -89,13 +89,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;
 }
@@ -177,6 +177,7 @@ int32_t mnodeInitDnodes() {
 
 void mnodeCleanupDnodes() {
   sdbCloseTable(tsDnodeSdb);
+  tsDnodeSdb = NULL;
 }
 
 void *mnodeGetNextDnode(void *pIter, SDnodeObj **pDnode) { 
@@ -357,7 +358,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);

src/mnode/src/mnodeMain.c

@@ -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();
 }
 

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;
@@ -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;
+  }
 
- static int32_t sdbForwardToPeer(SWalHead *pHead) {
-  if (tsSdbObj.sync == NULL) return TSDB_CODE_SUCCESS;
+  if (pMsg != NULL) {
+    sdbDebug("app:%p:%p, is confirmed and will do callback func", pMsg->rpcMsg.ahandle, pMsg);
+  }
 
-  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;
+  if (pOper->cb != NULL) {
+    pOper->retCode = (*pOper->cb)(pMsg, pOper->retCode);
+  }
+
+  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;
@@ -379,7 +385,6 @@ void sdbCleanUp() {
     tsSdbObj.wal = NULL;
   }
   
-  sem_destroy(&tsSdbObj.sem);
   pthread_mutex_destroy(&tsSdbObj.mutex);
 }
 
@@ -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), version:%" PRId64 " too large, sdb version:%" 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), version:%" PRId64 " too large, sdb version:%" 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 version:%" PRId64, pTable->tableName,
+               tstrerror(syncCode), sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
+    } else if (syncCode > 0) {
+      sdbDebug("table:%s, forward request is sent, action:%s record:%s version:%" PRId64, pTable->tableName,
+               sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
+    } else {
+      sdbTrace("table:%s, no need to send fwd request, action:%s record:%s version:%" PRId64, pTable->tableName,
+               sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
+    }
+    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 version:%" 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,20 @@ 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 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);
+        }
       } 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 (pOper && code <= 0) pOper->retCode = code;
     }
 
     walFsync(tsSdbObj.wal);
@@ -965,25 +976,17 @@ 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) {
+        syncConfirmForward(tsSdbObj.sync, pHead->version, TSDB_CODE_SUCCESS);
+        taosFreeQitem(item);
+      } else {
+        taosFreeQitem(item);
       }
-      taosFreeQitem(item);
     }
   }
 

src/mnode/src/mnodeTable.c

@@ -367,6 +367,7 @@ static int32_t mnodeInitChildTables() {
 
 static void mnodeCleanupChildTables() {
   sdbCloseTable(tsChildTableSdb);
+  tsChildTableSdb = NULL;
 }
 
 static void mnodeAddTableIntoStable(SSuperTableObj *pStable, SChildTableObj *pCtable) {
@@ -545,6 +546,7 @@ static int32_t mnodeInitSuperTables() {
 
 static void mnodeCleanupSuperTables() {
   sdbCloseTable(tsSuperTableSdb);
+  tsSuperTableSdb = NULL;
 }
 
 int32_t mnodeInitTables() {
@@ -774,9 +776,15 @@ static int32_t mnodeProcessTableMetaMsg(SMnodeMsg *pMsg) {
 
 static int32_t mnodeCreateSuperTableCb(SMnodeMsg *pMsg, int32_t code) {
   SSuperTableObj *pTable = (SSuperTableObj *)pMsg->pTable;
-  if (pTable != NULL) {
-    mLInfo("app:%p:%p, stable:%s, is created in sdb, result:%s", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId,
-            tstrerror(code));
+  assert(pTable);
+
+  if (code == TSDB_CODE_SUCCESS) {
+    mLInfo("stable:%s, is created in sdb", pTable->info.tableId);
+  } else {
+    mError("app:%p:%p, stable:%s, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pTable->info.tableId,
+           tstrerror(code));
+    SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pTable, .table = tsSuperTableSdb};
+    sdbDeleteRow(&desc);
   }
 
   return code;
@@ -1552,10 +1560,16 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
   SChildTableObj *pTable = (SChildTableObj *)pMsg->pTable;
   assert(pTable);
 
-  mDebug("app:%p:%p, table:%s, created in mnode, vgId:%d sid:%d, uid:%" PRIu64 ", result:%s", pMsg->rpcMsg.ahandle,
-         pMsg, pTable->info.tableId, pTable->vgId, pTable->sid, pTable->uid, tstrerror(code));
-
-  if (code != TSDB_CODE_SUCCESS) return code;
+  if (code == TSDB_CODE_SUCCESS) {
+    mDebug("app:%p:%p, table:%s, created in mnode, vgId:%d sid:%d, uid:%" PRIu64 ", result:%s", pMsg->rpcMsg.ahandle,
+           pMsg, pTable->info.tableId, pTable->vgId, pTable->sid, pTable->uid, tstrerror(code));
+  } else {
+    mError("app:%p:%p, table:%s, failed to create table sid:%d, uid:%" PRIu64 ", reason:%s", pMsg->rpcMsg.ahandle, pMsg,
+           pTable->info.tableId, pTable->sid, pTable->uid, tstrerror(code));
+    SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pTable, .table = tsChildTableSdb};
+    sdbDeleteRow(&desc);
+    return code;
+  }
 
   SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont;
   SMDCreateTableMsg *pMDCreate = mnodeBuildCreateChildTableMsg(pCreate, pTable);

src/mnode/src/mnodeUser.c

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

src/mnode/src/mnodeVgroup.c

@@ -140,6 +140,15 @@ static int32_t mnodeVgroupActionUpdate(SSdbOper *pOper) {
       }
       mnodeDecDnodeRef(pDnode);
     }
+
+    free(pNew);
+  }
+
+
+  // reset vgid status on vgroup changed
+  mDebug("vgId:%d, reset sync status to unsynced", pVgroup->vgId);
+  for (int32_t v = 0; v < pVgroup->numOfVnodes; ++v) {
+    pVgroup->vnodeGid[v].role = TAOS_SYNC_ROLE_UNSYNCED;
   }
 
   mnodeDecVgroupRef(pVgroup);
@@ -273,6 +282,7 @@ void mnodeUpdateVgroupStatus(SVgObj *pVgroup, SDnodeObj *pDnode, SVnodeLoad *pVl
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
     SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
     if (pVgid->pDnode == pDnode) {
+      mTrace("dnode:%d, receive status from dnode, vgId:%d status is %d", pVgroup->vgId, pDnode->dnodeId, pVgid->role);
       pVgid->role = pVload->role;
       if (pVload->role == TAOS_SYNC_ROLE_MASTER) {
         pVgroup->inUse = i;
@@ -436,17 +446,23 @@ void *mnodeGetNextVgroup(void *pIter, SVgObj **pVgroup) {
 }
 
 static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
+  SVgObj *pVgroup = pMsg->pVgroup;
+  SDbObj *pDb = pMsg->pDb;
+  assert(pVgroup);
+
   if (code != TSDB_CODE_SUCCESS) {
-    pMsg->pVgroup = NULL;
+    mError("app:%p:%p, vgId:%d, failed to create in sdb, reason:%s", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
+           tstrerror(code));
+    SSdbOper desc = {.type = SDB_OPER_GLOBAL, .pObj = pVgroup, .table = tsVgroupSdb};
+    sdbDeleteRow(&desc);
     return code;
   }
 
-  SVgObj *pVgroup = pMsg->pVgroup;
-  SDbObj *pDb = pMsg->pDb;
-
-  mInfo("vgId:%d, is created in mnode, db:%s replica:%d", pVgroup->vgId, pDb->name, pVgroup->numOfVnodes);
+  mInfo("app:%p:%p, vgId:%d, is created in mnode, db:%s replica:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId,
+        pDb->name, pVgroup->numOfVnodes);
   for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
-    mInfo("vgId:%d, index:%d, dnode:%d", pVgroup->vgId, i, pVgroup->vnodeGid[i].dnodeId);
+    mInfo("app:%p:%p, vgId:%d, index:%d, dnode:%d", pMsg->rpcMsg.ahandle, pMsg, pVgroup->vgId, i,
+          pVgroup->vnodeGid[i].dnodeId);
   }
 
   mnodeIncVgroupRef(pVgroup);
@@ -510,6 +526,7 @@ void mnodeDropVgroup(SVgObj *pVgroup, void *ahandle) {
 
 void mnodeCleanupVgroups() {
   sdbCloseTable(tsVgroupSdb);
+  tsVgroupSdb = NULL;
 }
 
 int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
@@ -791,9 +808,9 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
   if (rpcMsg->ahandle == NULL) return;
 
   SMnodeMsg *mnodeMsg = rpcMsg->ahandle;
-  mnodeMsg->received++;
+  atomic_add_fetch_8(&mnodeMsg->received, 1);
   if (rpcMsg->code == TSDB_CODE_SUCCESS) {
-    mnodeMsg->successed++;
+    atomic_add_fetch_8(&mnodeMsg->successed, 1);
   } else {
     mnodeMsg->code = rpcMsg->code;
   }

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, 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);
 
   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",

src/plugins/http/src/httpSql.c

@@ -166,8 +166,8 @@ void httpProcessMultiSql(HttpContext *pContext) {
   HttpSqlCmd *cmd = multiCmds->cmds + multiCmds->pos;
 
   char *sql = httpGetCmdsString(pContext, cmd->sql);
-  httpTraceDump("context:%p, fd:%d, ip:%s, user:%s, process pos:%d, start query, sql:%s", pContext, pContext->fd,
-           pContext->ipstr, pContext->user, multiCmds->pos, sql);
+  httpTraceL("context:%p, fd:%d, ip:%s, user:%s, process pos:%d, start query, sql:%s", pContext, pContext->fd,
+             pContext->ipstr, pContext->user, multiCmds->pos, sql);
   taosNotePrintHttp(sql);
   taos_query_a(pContext->session->taos, sql, httpProcessMultiSqlCallBack, (void *)pContext);
 }
@@ -306,8 +306,8 @@ void httpProcessSingleSqlCmd(HttpContext *pContext) {
     return;
   }
 
-  httpTraceDump("context:%p, fd:%d, ip:%s, user:%s, start query, sql:%s", pContext, pContext->fd, pContext->ipstr,
-           pContext->user, sql);
+  httpTraceL("context:%p, fd:%d, ip:%s, user:%s, start query, sql:%s", pContext, pContext->fd, pContext->ipstr,
+            pContext->user, sql);
   taosNotePrintHttp(sql);
   taos_query_a(pSession->taos, sql, httpProcessSingleSqlCallBack, (void *)pContext);
 }

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

src/plugins/mqtt/inc/mqttLog.h

@@ -27,7 +27,4 @@ extern int32_t mqttDebugFlag;
 #define mqttDebug(...) { if (mqttDebugFlag & DEBUG_DEBUG) { taosPrintLog("MQT DEBUG ", mqttDebugFlag, __VA_ARGS__); }}
 #define mqttTrace(...) { if (mqttDebugFlag & DEBUG_TRACE) { taosPrintLog("MQT TRACE ", mqttDebugFlag, __VA_ARGS__); }}
 
-#define mqttDebugDump(...) { if (mqttDebugFlag & DEBUG_DEBUG) { taosPrintLongString("MQT DEBUG ", mqttDebugFlag, __VA_ARGS__); }}
-#define mqttTraceDump(...) { if (mqttDebugFlag & DEBUG_TRACE) { taosPrintLongString("MQT DEBUG ", mqttDebugFlag, __VA_ARGS__); }}
-
 #endif

src/query/inc/qExecutor.h

@@ -154,6 +154,7 @@ typedef struct SQuery {
 } SQuery;
 
 typedef struct SQueryRuntimeEnv {
+  jmp_buf              env;
   SResultInfo*         resultInfo;       // todo refactor to merge with SWindowResInfo
   SQuery*              pQuery;
   SQLFunctionCtx*      pCtx;
@@ -169,6 +170,8 @@ typedef struct SQueryRuntimeEnv {
   void*                pSecQueryHandle;  // another thread for
   bool                 stableQuery;      // super table query or not
   bool                 topBotQuery;      // false
+  bool                 groupbyNormalCol; // denote if this is a groupby normal column query
+  bool                 hasTagResults;    // if there are tag values in final result or not
   int32_t              prevGroupId;      // previous executed group id
   SDiskbasedResultBuf* pResultBuf;       // query result buffer based on blocked-wised disk file
 } SQueryRuntimeEnv;
@@ -197,8 +200,10 @@ typedef struct SQInfo {
    */
   int32_t          tableIndex;
   int32_t          numOfGroupResultPages;
-  _qinfo_free_fn_t freeFn;
-  jmp_buf          env;
+  _qinfo_free_fn_t freeFn;  //todo remove it
+
+  void*     pBuf; // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
+
 } SQInfo;
 
 #endif  // TDENGINE_QUERYEXECUTOR_H

src/query/inc/qast.h

@@ -16,16 +16,16 @@
 #ifndef TDENGINE_TAST_H
 #define TDENGINE_TAST_H
 
-#include <tbuffer.h>
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#include <tskiplist.h>
 #include "os.h"
 
 #include "taosmsg.h"
 #include "taosdef.h"
+#include "tskiplist.h"
+#include "tbuffer.h"
 #include "tvariant.h"
 
 struct tExprNode;
@@ -75,10 +75,6 @@ typedef struct tExprNode {
   };
 } tExprNode;
 
-void tSQLBinaryExprFromString(tExprNode **pExpr, SSchema *pSchema, int32_t numOfCols, char *src, int32_t len);
-
-void tSQLBinaryExprToString(tExprNode *pExpr, char *dst, int32_t *len);
-
 void tExprTreeDestroy(tExprNode **pExprs, void (*fp)(void*));
 
 void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param);
@@ -86,12 +82,9 @@ void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, S
 void tExprTreeCalcTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
                                 char *(*cb)(void *, const char*, int32_t));
 
-// todo refactor: remove it
-void tSQLBinaryExprTrv(tExprNode *pExprs, SArray* res);
-
 uint8_t getBinaryExprOptr(SSQLToken *pToken);
 
-void       tExprNodeDestroy(tExprNode *pNode, void (*fp)(void *));
+void tExprNodeDestroy(tExprNode *pNode, void (*fp)(void *));
 void exprTreeToBinary(SBufferWriter* bw, tExprNode* pExprTree);
 
 tExprNode* exprTreeFromBinary(const void* data, size_t size);

src/query/inc/qfill.h

@@ -60,8 +60,6 @@ typedef struct SPoint {
   void *  val;
 } SPoint;
 
-int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, char timeUnit, int16_t precision);
-
 SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
                             int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
                             SFillColInfo* pFillCol);

src/query/src/qUtil.c

@@ -32,7 +32,6 @@ int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRun
   pWindowResInfo->threshold = threshold;
   
   pWindowResInfo->type = type;
-  
   _hash_fn_t fn = taosGetDefaultHashFunction(type);
   pWindowResInfo->hashList = taosHashInit(threshold, fn, false);
   
@@ -54,7 +53,8 @@ void destroyTimeWindowRes(SWindowResult *pWindowRes, int32_t nOutputCols) {
   if (pWindowRes == NULL) {
     return;
   }
-  
+
+  // TODO opt malloc strategy
   for (int32_t i = 0; i < nOutputCols; ++i) {
     free(pWindowRes->resultInfo[i].interResultBuf);
   }
@@ -180,19 +180,33 @@ void closeAllTimeWindow(SWindowResInfo *pWindowResInfo) {
 
 /*
  * remove the results that are not the FIRST time window that spreads beyond the
- * the last qualified time stamp in case of sliding query, which the sliding time is not equalled to the interval time
+ * the last qualified time stamp in case of sliding query, which the sliding time is not equalled to the interval time.
+ * NOTE: remove redundant, only when the result set order equals to traverse order
  */
 void removeRedundantWindow(SWindowResInfo *pWindowResInfo, TSKEY lastKey, int32_t order) {
   assert(pWindowResInfo->size >= 0 && pWindowResInfo->capacity >= pWindowResInfo->size);
-  
+  if (pWindowResInfo->size <= 1) {
+    return;
+  }
+
+  // get the result order
+  int32_t resultOrder = (pWindowResInfo->pResult[0].window.skey < pWindowResInfo->pResult[1].window.skey)? 1:-1;
+
+  if (order != resultOrder) {
+    return;
+  }
+
   int32_t i = 0;
-  while (i < pWindowResInfo->size &&
-      ((pWindowResInfo->pResult[i].window.ekey < lastKey && order == QUERY_ASC_FORWARD_STEP) ||
-          (pWindowResInfo->pResult[i].window.skey > lastKey && order == QUERY_DESC_FORWARD_STEP))) {
-    ++i;
+  if (order == QUERY_ASC_FORWARD_STEP) {
+    while (i < pWindowResInfo->size && (pWindowResInfo->pResult[i].window.ekey < lastKey)) {
+      ++i;
+    }
+  } else if (order == QUERY_DESC_FORWARD_STEP) {
+    while (i < pWindowResInfo->size && (pWindowResInfo->pResult[i].window.skey > lastKey)) {
+      ++i;
+    }
   }
-  
-  //  assert(i < pWindowResInfo->size);
+
   if (i < pWindowResInfo->size) {
     pWindowResInfo->size = (i + 1);
   }

src/query/src/qextbuffer.c

@@ -118,7 +118,7 @@ static bool tExtMemBufferAlloc(tExtMemBuffer *pMemBuffer) {
    * To flush data to disk to accommodate more data
    */
   if (pMemBuffer->numOfInMemPages > 0 && pMemBuffer->numOfInMemPages == pMemBuffer->inMemCapacity) {
-    if (!tExtMemBufferFlush(pMemBuffer)) {
+    if (tExtMemBufferFlush(pMemBuffer) != 0) {
       return false;
     }
   }
@@ -268,6 +268,7 @@ int32_t tExtMemBufferFlush(tExtMemBuffer *pMemBuffer) {
     size_t retVal = fwrite((char *)&(first->item), pMemBuffer->pageSize, 1, pMemBuffer->file);
     if (retVal <= 0) {  // failed to write to buffer, may be not enough space
       ret = TAOS_SYSTEM_ERROR(errno);
+      return ret;
     }
 
     pMemBuffer->fileMeta.numOfElemsInFile += first->item.num;

src/query/src/qfill.c

@@ -22,41 +22,6 @@
 
 #define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
 
-int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, char timeUnit, int16_t precision) {
-  if (slidingTime == 0) {
-    return startTime;
-  }
-
-  if (timeUnit == 'a' || timeUnit == 'm' || timeUnit == 's' || timeUnit == 'h') {
-    return (startTime / slidingTime) * slidingTime;
-  } else {
-    /*
-     * 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 dynamically decide the start time of a day, move to common module
-     */
-
-    // 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;
-
-    int64_t revStartime = (startTime / slidingTime) * slidingTime + timezone * t;
-    int64_t revEndtime = revStartime + slidingTime - 1;
-    if (revEndtime < startTime) {
-      revStartime += slidingTime;
-    }
-
-    return revStartime;
-  }
-}
-
 SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
     int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType, SFillColInfo* pFillCol) {
   if (fillType == TSDB_FILL_NONE) {
@@ -128,7 +93,7 @@ static TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int64_t timeInterva
   if (order == TSDB_ORDER_ASC) {
     return ekey;
   } else {
-    return taosGetIntervalStartTimestamp(ekey, timeInterval, slidingTimeUnit, precision);
+    return taosGetIntervalStartTimestamp(ekey, timeInterval, timeInterval, slidingTimeUnit, precision);
   }
 }
 

src/rpc/inc/rpcLog.h

@@ -31,9 +31,7 @@ extern int32_t tscEmbedded;
 #define tInfo(...)  { if (rpcDebugFlag & DEBUG_INFO)  { taosPrintLog("RPC INFO  ", tscEmbedded ? 255 : rpcDebugFlag, __VA_ARGS__); }}
 #define tDebug(...) { if (rpcDebugFlag & DEBUG_DEBUG) { taosPrintLog("RPC DEBUG ", rpcDebugFlag, __VA_ARGS__); }}
 #define tTrace(...) { if (rpcDebugFlag & DEBUG_TRACE) { taosPrintLog("RPC TRACE ", rpcDebugFlag, __VA_ARGS__); }}
-
-#define tDebugDump(x, y) { if (rpcDebugFlag & DEBUG_DEBUG)  { taosDumpData((unsigned char *)x, y); }}
-#define tTraceDump(x, y) { if (rpcDebugFlag & DEBUG_TRACE)  { taosDumpData((unsigned char *)x, y); }}
+#define tDump(x, y) { if (rpcDebugFlag & DEBUG_DUMP)  { taosDumpData((unsigned char *)x, y); }}
 
 #ifdef __cplusplus
 }

src/rpc/test/rclient.c

@@ -156,6 +156,7 @@ int main(int argc, char *argv[]) {
   }
 
   tInfo("client is initialized");
+  tInfo("threads:%d msgSize:%d requests:%d", appThreads, msgSize, numOfReqs);
 
   gettimeofday(&systemTime, NULL);
   startTime = systemTime.tv_sec*1000000 + systemTime.tv_usec;

src/util/inc/tlog.h

@@ -26,6 +26,7 @@ extern "C" {
 #define DEBUG_INFO  DEBUG_WARN
 #define DEBUG_DEBUG 4U
 #define DEBUG_TRACE 8U
+#define DEBUG_DUMP  16U
 
 #define DEBUG_SCREEN 64U
 #define DEBUG_FILE   128U

src/query/src/tlosertree.c → src/util/src/tlosertree.c

@@ -13,10 +13,10 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "tlosertree.h"
 #include "os.h"
 #include "taosmsg.h"
-#include "tlosertree.h"
-#include "queryLog.h"
+#include "tulog.h"
 
 // set initial value for loser tree
 void tLoserTreeInit(SLoserTreeInfo* pTree) {
@@ -45,7 +45,7 @@ uint32_t tLoserTreeCreate(SLoserTreeInfo** pTree, int32_t numOfEntries, void* pa
 
   *pTree = (SLoserTreeInfo*)calloc(1, sizeof(SLoserTreeInfo) + sizeof(SLoserTreeNode) * totalEntries);
   if ((*pTree) == NULL) {
-    qError("allocate memory for loser-tree failed. reason:%s", strerror(errno));
+    uError("allocate memory for loser-tree failed. reason:%s", strerror(errno));
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 

tests/pytest/alter/alter_table_crash.py

@@ -78,5 +78,6 @@ class TDTestCase:
         tdSql.close()
         tdLog.success("%s successfully executed" % __file__)
 
+
 tdCases.addWindows(__file__, TDTestCase())
 tdCases.addLinux(__file__, TDTestCase())