Merge branch 'develop' into test/testcase

documentation20/webdocs/markdowndocs/administrator-ch.md

@@ -95,6 +95,7 @@ TDengine系统后台服务由taosd提供，可以在配置文件taos.cfg里修
 - logKeepDays：日志文件的最长保存时间。大于0时，日志文件会被重命名为taosdlog.xxx，其中xxx为日志文件最后修改的时间戳，单位为秒。默认值：0天。
 - maxSQLLength：单条SQL语句允许最长限制。默认值：65380字节。
 - telemetryReporting: 是否允许 TDengine 采集和上报基本使用信息，0表示不允许，1表示允许。 默认值：1。
+- stream: 是否启用连续查询（流计算功能），0表示不允许，1表示允许。 默认值：1。
 
 **注意：**对于端口，TDengine会使用从serverPort起13个连续的TCP和UDP端口号，请务必在防火墙打开。因此如果是缺省配置，需要打开从6030都6042共13个端口，而且必须TCP和UDP都打开。
 

documentation20/webdocs/markdowndocs/connector-ch.md

@@ -142,7 +142,7 @@ C/C++的API类似于MySQL的C API。应用程序使用时，需要包含TDengine
   获取最近一次API调用失败的原因，返回值为错误代码。
 
 
-**注意**：对于单个数据库连接，在同一时刻只能有一个线程使用该连接调用API，否则会有未定义的行为出现并可能导致客户端crash。客户端应用可以通过建立多个连接进行多线程的数据写入或查询处理。
+**注意**：对于每个数据库应用，2.0及以上版本 TDengine 推荐只建立一个连接。同时在应用中将该连接 (TAOS*) 结构体传递到不同的线程共享使用。基于 TAOS 结构体发出的查询、写入等操作具有多线程安全性。C 语言的连接器可以按照需求动态建立面向数据库的新连接（该过程对用户不可见），同时建议只有在程序最后退出的时候才调用 taos_close 关闭连接。
 
 
 ### 异步查询API

packaging/cfg/taos.cfg

@@ -236,7 +236,7 @@
 # httpDebugFlag         131
 
 # debug flag for monitor
-# monitorDebugFlag      131
+# monDebugFlag          131
 
 # debug flag for query
 # qDebugflag            131
@@ -251,7 +251,7 @@
 # cqDebugFlag           131
 
 # enable/disable recording the SQL in taos client
-# tscEnableRecordSql    0
+# enableRecordSql    0
 
 # generate core file when service crash
 # enableCoreFile        1
@@ -260,4 +260,10 @@
 # maxBinaryDisplayWidth 30
 
 # enable/disable telemetry reporting
-# telemetryReporting    1
+# telemetryReporting    1
\ No newline at end of file
+
+# enable/disable stream (continuous query)
+# stream                1
+
+# only 50% CPU resources will be used in query processing
+# halfCoresForQuery     0

src/balance/inc/bnInt.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_BALANCE_INT_H
+#define TDENGINE_BALANCE_INT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "mnodeInt.h"
+#include "mnodeDef.h"
+#include "mnodeDnode.h"
+
+typedef struct {
+  int32_t     size;
+  int32_t     maxSize;
+  SDnodeObj **list;
+} SBnDnodes;
+
+typedef struct {
+  void *          timer;
+  bool            stop;
+  pthread_mutex_t mutex;
+  pthread_cond_t  cond;
+  pthread_t       thread;
+} SBnThread;
+
+typedef struct {
+  pthread_mutex_t mutex;
+} SBnMgmt;
+
+int32_t bnInit();
+void    bnCleanUp();
+bool    bnStart();
+void    bnCheckStatus();
+void    bnCheckModules();
+
+extern SBnDnodes tsBnDnodes;
+extern void *tsMnodeTmr;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/balance/inc/bnScore.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_BALANCE_SCORE_H
+#define TDENGINE_BALANCE_SCORE_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "bnInt.h"
+
+void  bnInitDnodes();
+void  bnCleanupDnodes();
+void  bnAccquireDnodes();
+void  bnReleaseDnodes();
+float bnTryCalcDnodeScore(SDnodeObj *pDnode, int32_t extraVnode);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/balance/inc/bnThread.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef TDENGINE_BALANCE_THREAD_H
+#define TDENGINE_BALANCE_THREAD_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "bnInt.h"
+
+int32_t bnInitThread();
+void    bnCleanupThread();
+void    bnNotify();
+void    bnStartTimer(int64_t mseconds);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/balance/src/bnScore.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "os.h"
+#include "tglobal.h"
+#include "mnodeShow.h"
+#include "mnodeUser.h"
+#include "bnScore.h"
+
+SBnDnodes tsBnDnodes;
+
+static int32_t bnGetScoresMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
+static int32_t bnRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn);
+
+static int32_t bnCalcCpuScore(SDnodeObj *pDnode) {
+  if (pDnode->cpuAvgUsage < 80)
+    return 0;
+  else if (pDnode->cpuAvgUsage < 90)
+    return 10;
+  else
+    return 50;
+}
+
+static int32_t bnCalcMemoryScore(SDnodeObj *pDnode) {
+  if (pDnode->memoryAvgUsage < 80)
+    return 0;
+  else if (pDnode->memoryAvgUsage < 90)
+    return 10;
+  else
+    return 50;
+}
+
+static int32_t bnCalcDiskScore(SDnodeObj *pDnode) {
+  if (pDnode->diskAvgUsage < 80)
+    return 0;
+  else if (pDnode->diskAvgUsage < 90)
+    return 10;
+  else
+    return 50;
+}
+
+static int32_t bnCalcBandScore(SDnodeObj *pDnode) {
+  if (pDnode->bandwidthUsage < 30)
+    return 0;
+  else if (pDnode->bandwidthUsage < 80)
+    return 10;
+  else
+    return 50;
+}
+
+static float bnCalcModuleScore(SDnodeObj *pDnode) {
+  if (pDnode->numOfCores <= 0) return 0;
+  if (pDnode->isMgmt) {
+    return (float)tsMnodeEqualVnodeNum / pDnode->numOfCores;
+  }
+  return 0;
+}
+
+static float bnCalcVnodeScore(SDnodeObj *pDnode, int32_t extra) {
+  if (pDnode->status == TAOS_DN_STATUS_DROPPING || pDnode->status == TAOS_DN_STATUS_OFFLINE) return 100000000;
+  if (pDnode->numOfCores <= 0) return 0;
+  return (float)(pDnode->openVnodes + extra) / pDnode->numOfCores;
+}
+
+/**
+ * calc singe score, such as cpu/memory/disk/bandwitdh/vnode
+ * 1. get the score config
+ * 2. if the value is out of range, use border data
+ * 3. otherwise use interpolation method
+ **/
+static void bnCalcDnodeScore(SDnodeObj *pDnode) {
+  pDnode->score = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) +
+                  bnCalcBandScore(pDnode) + bnCalcModuleScore(pDnode) + bnCalcVnodeScore(pDnode, 0) +
+                  pDnode->customScore;
+}
+
+float bnTryCalcDnodeScore(SDnodeObj *pDnode, int32_t extra) {
+  int32_t systemScore = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) +
+                        bnCalcBandScore(pDnode);
+  float moduleScore = bnCalcModuleScore(pDnode);
+  float vnodeScore = bnCalcVnodeScore(pDnode, extra);
+
+  float score = systemScore + moduleScore + vnodeScore +  pDnode->customScore;
+  return score;
+}
+
+void bnInitDnodes() {
+  mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_SCORES, bnGetScoresMeta);
+  mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_SCORES, bnRetrieveScores);
+  mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_SCORES, mnodeCancelGetNextDnode);
+
+  memset(&tsBnDnodes, 0, sizeof(SBnDnodes));
+  tsBnDnodes.maxSize = 16;
+  tsBnDnodes.list = calloc(tsBnDnodes.maxSize, sizeof(SDnodeObj *));
+}
+
+void bnCleanupDnodes() {
+  if (tsBnDnodes.list != NULL) {
+    free(tsBnDnodes.list);
+    tsBnDnodes.list = NULL;
+  }
+}
+
+static void bnCheckDnodesSize(int32_t dnodesNum) {
+  if (tsBnDnodes.maxSize <= dnodesNum) {
+    tsBnDnodes.maxSize = dnodesNum * 2;
+    tsBnDnodes.list = realloc(tsBnDnodes.list, tsBnDnodes.maxSize * sizeof(SDnodeObj *));
+  }
+}
+
+void bnAccquireDnodes() {
+  int32_t dnodesNum = mnodeGetDnodesNum();
+  bnCheckDnodesSize(dnodesNum);
+
+  void *     pIter = NULL;
+  SDnodeObj *pDnode = NULL;
+  int32_t    dnodeIndex = 0;
+
+  while (1) {
+    if (dnodeIndex >= dnodesNum) {
+      mnodeCancelGetNextDnode(pIter);
+      break;
+    }
+
+    pIter = mnodeGetNextDnode(pIter, &pDnode);
+    if (pDnode == NULL) break;
+    if (pDnode->status == TAOS_DN_STATUS_OFFLINE) {
+      mnodeDecDnodeRef(pDnode);
+      continue;
+    }
+
+    bnCalcDnodeScore(pDnode);
+    
+    int32_t orderIndex = dnodeIndex;
+    for (; orderIndex > 0; --orderIndex) {
+      if (pDnode->score > tsBnDnodes.list[orderIndex - 1]->score) {
+        break;
+      }
+      tsBnDnodes.list[orderIndex] = tsBnDnodes.list[orderIndex - 1];
+    }
+    tsBnDnodes.list[orderIndex] = pDnode;
+    dnodeIndex++;
+  }
+
+  tsBnDnodes.size = dnodeIndex;
+}
+
+void bnReleaseDnodes() {
+  for (int32_t i = 0; i < tsBnDnodes.size; ++i) {
+    SDnodeObj *pDnode = tsBnDnodes.list[i];
+    if (pDnode != NULL) {
+      mnodeDecDnodeRef(pDnode);
+    }
+  }
+}
+
+static int32_t bnGetScoresMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
+  SUserObj *pUser = mnodeGetUserFromConn(pConn);
+  if (pUser == NULL) return 0;
+
+  if (strcmp(pUser->pAcct->user, "root") != 0) {
+    mnodeDecUserRef(pUser);
+    return TSDB_CODE_MND_NO_RIGHTS;
+  }
+
+  int32_t  cols = 0;
+  SSchema *pSchema = pMeta->schema;
+
+  pShow->bytes[cols] = 2;
+  pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
+  strcpy(pSchema[cols].name, "id");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
+  strcpy(pSchema[cols].name, "system scores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
+  strcpy(pSchema[cols].name, "custom scores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
+  strcpy(pSchema[cols].name, "module scores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
+  strcpy(pSchema[cols].name, "vnode scores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_FLOAT;
+  strcpy(pSchema[cols].name, "total scores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_INT;
+  strcpy(pSchema[cols].name, "open vnodes");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 4;
+  pSchema[cols].type = TSDB_DATA_TYPE_INT;
+  strcpy(pSchema[cols].name, "cpu cores");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pShow->bytes[cols] = 18 + VARSTR_HEADER_SIZE;
+  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
+  strcpy(pSchema[cols].name, "balance state");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
+  pMeta->numOfColumns = htons(cols);
+  pShow->numOfColumns = cols;
+
+  pShow->offset[0] = 0;
+  for (int32_t i = 1; i < cols; ++i) {
+    pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
+  }
+
+  pShow->numOfRows = mnodeGetDnodesNum();
+  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
+  pShow->pIter = NULL;
+
+  mnodeDecUserRef(pUser);
+
+  return 0;
+}
+
+static int32_t bnRetrieveScores(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
+  int32_t    numOfRows = 0;
+  SDnodeObj *pDnode = NULL;
+  char *     pWrite;
+  int32_t    cols = 0;
+
+  while (numOfRows < rows) {
+    pShow->pIter = mnodeGetNextDnode(pShow->pIter, &pDnode);
+    if (pDnode == NULL) break;
+
+    int32_t systemScore = bnCalcCpuScore(pDnode) + bnCalcMemoryScore(pDnode) + bnCalcDiskScore(pDnode) + bnCalcBandScore(pDnode);
+    float moduleScore = bnCalcModuleScore(pDnode);
+    float vnodeScore = bnCalcVnodeScore(pDnode, 0);
+
+    cols = 0;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int16_t *)pWrite = pDnode->dnodeId;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(float *)pWrite = systemScore;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(float *)pWrite = pDnode->customScore;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(float *)pWrite = (int32_t)moduleScore;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(float *)pWrite = (int32_t)vnodeScore;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(float *)pWrite = (int32_t)(vnodeScore + moduleScore + pDnode->customScore + systemScore);
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int32_t *)pWrite = pDnode->openVnodes;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int32_t *)pWrite = pDnode->numOfCores;
+    cols++;
+
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    STR_TO_VARSTR(pWrite, mnodeGetDnodeStatusStr(pDnode->status));
+    cols++;
+
+    numOfRows++;
+    mnodeDecDnodeRef(pDnode);
+  }
+
+  mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
+  pShow->numOfReads += numOfRows;
+  return numOfRows;
+}

src/balance/src/bnThread.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "os.h"
+#include "ttimer.h"
+#include "tglobal.h"
+#include "mnodeSdb.h"
+#include "bnThread.h"
+
+static SBnThread tsBnThread;
+
+static void *bnThreadFunc(void *arg) {
+  while (1) {
+    pthread_mutex_lock(&tsBnThread.mutex);
+    if (tsBnThread.stop) {
+      pthread_mutex_unlock(&tsBnThread.mutex);
+      break;
+    }
+
+    pthread_cond_wait(&tsBnThread.cond, &tsBnThread.mutex);
+    bool updateSoon = bnStart();
+    bnStartTimer(updateSoon ? 1000 : -1);
+    pthread_mutex_unlock(&(tsBnThread.mutex));
+  }
+
+  mDebug("balance thread is stopped");
+  return NULL;
+}
+
+int32_t bnInitThread() {
+  memset(&tsBnThread, 0, sizeof(SBnThread));
+  tsBnThread.stop = false;
+  pthread_mutex_init(&tsBnThread.mutex, NULL);
+  pthread_cond_init(&tsBnThread.cond, NULL);
+
+  pthread_attr_t thattr;
+  pthread_attr_init(&thattr);
+  pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
+  int32_t ret = pthread_create(&tsBnThread.thread, &thattr, bnThreadFunc, NULL);
+  pthread_attr_destroy(&thattr);
+
+  if (ret != 0) {
+    mError("failed to create balance thread since %s", strerror(errno));
+    return -1;
+  }
+
+  bnStartTimer(2000);
+  mDebug("balance thread is created");
+  return 0;
+}
+
+void bnCleanupThread() {
+  mDebug("balance thread will be cleanup");
+
+  if (tsBnThread.timer != NULL) {
+    taosTmrStopA(&tsBnThread.timer);
+    tsBnThread.timer = NULL;
+    mDebug("stop balance timer");
+  }
+
+  pthread_mutex_lock(&tsBnThread.mutex);
+  tsBnThread.stop = true;
+  pthread_cond_signal(&tsBnThread.cond);
+  pthread_mutex_unlock(&(tsBnThread.mutex));
+  pthread_join(tsBnThread.thread, NULL);
+
+  pthread_cond_destroy(&tsBnThread.cond);
+  pthread_mutex_destroy(&tsBnThread.mutex);
+}
+
+static void bnPostSignal() {
+  if (tsBnThread.stop) return;
+
+  pthread_mutex_lock(&tsBnThread.mutex);
+  pthread_cond_signal(&tsBnThread.cond);
+  pthread_mutex_unlock(&(tsBnThread.mutex));
+}
+
+/*
+ * once sdb work as mater, then tsAccessSquence reset to zero
+ * increase tsAccessSquence every balance interval
+ */
+
+static void bnProcessTimer(void *handle, void *tmrId) {
+  if (!sdbIsMaster()) return;
+  if (tsBnThread.stop) return;
+
+  tsBnThread.timer = NULL;
+  tsAccessSquence++;
+
+  bnCheckStatus();
+  bnStartTimer(-1);
+
+  if (handle == NULL) {
+    if (tsAccessSquence % tsBalanceInterval == 0) {
+      mDebug("balance function is scheduled by timer");
+      bnPostSignal();
+    }
+  } else {
+    int64_t mseconds = (int64_t)handle;
+    mDebug("balance function is scheduled by event for %" PRId64 " mseconds arrived", mseconds);
+    bnPostSignal();
+  }
+}
+
+void bnStartTimer(int64_t mseconds) {
+  if (tsBnThread.stop) return;
+
+  bool updateSoon = (mseconds != -1);
+  if (updateSoon) {
+    taosTmrReset(bnProcessTimer, mseconds, (void *)mseconds, tsMnodeTmr, &tsBnThread.timer);
+  } else {
+    taosTmrReset(bnProcessTimer, tsStatusInterval * 1000, NULL, tsMnodeTmr, &tsBnThread.timer);
+  }
+}
+
+void bnNotify() {
+  bnStartTimer(500); 
+}

src/client/inc/tsclient.h

@@ -246,11 +246,14 @@ typedef struct SQueryInfo {
   int16_t          fillType;      // final result fill type
   int16_t          numOfTables;
   STableMetaInfo **pTableMetaInfo;
-  struct STSBuf *  tsBuf;
+  struct STSBuf   *tsBuf;
   int64_t *        fillVal;       // default value for fill
   char *           msg;           // pointer to the pCmd->payload to keep error message temporarily
   int64_t          clauseLimit;   // limit for current sub clause
+
   int64_t          prjOffset;     // offset value in the original sql expression, only applied at client side
+  int64_t          tableLimit;    // table limit in case of super table projection query + global order + limit
+
   int32_t          udColumnId;    // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
   int16_t          resColumnId;   // result column id
 } SQueryInfo;

src/client/src/tscFunctionImpl.c

@@ -426,8 +426,7 @@ static void count_function_f(SQLFunctionCtx *pCtx, int32_t index) {
   }
   
   SET_VAL(pCtx, 1, 1);
-  
-  *((int64_t *)pCtx->aOutputBuf) += 1;
+  *((int64_t *)pCtx->aOutputBuf) += pCtx->size;
   
   // do not need it actually
   SResultRowCellInfo *pInfo = GET_RES_INFO(pCtx);
@@ -3632,114 +3631,119 @@ static bool twa_function_setup(SQLFunctionCtx *pCtx) {
   return true;
 }
 
-static int32_t twa_function_impl(SQLFunctionCtx* pCtx, int32_t index, int32_t size) {
+static int32_t twa_function_impl(SQLFunctionCtx* pCtx, int32_t tsIndex, int32_t index, int32_t size) {
   int32_t notNullElems = 0;
   TSKEY *primaryKey = pCtx->ptsList;
   
   SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
 
   STwaInfo *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
   int32_t i = index;
+  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
 
   if (pCtx->start.key != INT64_MIN) {
-    assert(pCtx->start.key < primaryKey[index] && pInfo->lastKey == INT64_MIN);
+    assert((pCtx->start.key < primaryKey[tsIndex + i] && pCtx->order == TSDB_ORDER_ASC) ||
+               (pCtx->start.key > primaryKey[tsIndex + i] && pCtx->order == TSDB_ORDER_DESC));
 
-    pInfo->lastKey = primaryKey[index];
-    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, 0));
+    assert(pInfo->lastKey == INT64_MIN);
+
+    pInfo->lastKey = primaryKey[tsIndex + i];
+    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, index));
 
     pInfo->dOutput += ((pInfo->lastValue + pCtx->start.val) / 2) * (pInfo->lastKey - pCtx->start.key);
 
     pInfo->hasResult = DATA_SET_FLAG;
-    pInfo->win.skey  = pCtx->start.key;
+    pInfo->win.skey = pCtx->start.key;
     notNullElems++;
-    i += 1;
+    i += step;
   } else if (pInfo->lastKey == INT64_MIN) {
-    pInfo->lastKey = primaryKey[index];
-    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, 0));
+    pInfo->lastKey = primaryKey[tsIndex + i];
+    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, index));
 
     pInfo->hasResult = DATA_SET_FLAG;
-    pInfo->win.skey  = pInfo->lastKey;
+    pInfo->win.skey = pInfo->lastKey;
     notNullElems++;
-    i += 1;
+    i += step;
   }
 
   // calculate the value of
   switch(pCtx->inputType) {
     case TSDB_DATA_TYPE_TINYINT: {
-      int8_t *val = (int8_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      int8_t *val = (int8_t*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
     case TSDB_DATA_TYPE_SMALLINT: {
-      int16_t *val = (int16_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      int16_t *val = (int16_t*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
     case TSDB_DATA_TYPE_INT: {
-      int32_t *val = (int32_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      int32_t *val = (int32_t*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
     case TSDB_DATA_TYPE_BIGINT: {
-      int64_t *val = (int64_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      int64_t *val = (int64_t*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = (double) val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
     case TSDB_DATA_TYPE_FLOAT: {
-      float *val = (float*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      float *val = (float*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
     case TSDB_DATA_TYPE_DOUBLE: {
-      double *val = (double*) GET_INPUT_CHAR_INDEX(pCtx, index);
-      for (; i < size; i++) {
+      double *val = (double*) GET_INPUT_CHAR_INDEX(pCtx, 0);
+      for (; i < size && i >= 0; i += step) {
         if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
           continue;
         }
 
-        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i] - pInfo->lastKey);
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + tsIndex] - pInfo->lastKey);
         pInfo->lastValue = val[i];
-        pInfo->lastKey = primaryKey[i];
+        pInfo->lastKey = primaryKey[i + tsIndex];
       }
       break;
     }
@@ -3764,16 +3768,13 @@ static void twa_function(SQLFunctionCtx *pCtx) {
   STwaInfo *   pInfo = GET_ROWCELL_INTERBUF(pResInfo);
   
   // skip null value
-  int32_t i = 0;
+  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pCtx->order);
+  int32_t i = (pCtx->order == TSDB_ORDER_ASC)? 0:(pCtx->size - 1);
   while (pCtx->hasNull && i < pCtx->size && isNull((char *)data + pCtx->inputBytes * i, pCtx->inputType)) {
-    i++;
+    i += step;
   }
-  
-  if (i >= pCtx->size) {
-    return;
-  }
-  
-  int32_t notNullElems = twa_function_impl(pCtx, pCtx->startOffset, pCtx->size);
+
+  int32_t notNullElems = twa_function_impl(pCtx, pCtx->startOffset, i, pCtx->size);
   SET_VAL(pCtx, notNullElems, 1);
   
   if (notNullElems > 0) {
@@ -3791,11 +3792,136 @@ static void twa_function_f(SQLFunctionCtx *pCtx, int32_t index) {
     return;
   }
 
-  int32_t notNullElems = twa_function_impl(pCtx, index, 1);
+  int32_t notNullElems = 0;
+  TSKEY *primaryKey = pCtx->ptsList;
+
+  SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
+
+  STwaInfo *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+  int32_t i = pCtx->startOffset;
+  int32_t size = pCtx->size;
+
+  if (pCtx->start.key != INT64_MIN) {
+    assert(pInfo->lastKey == INT64_MIN);
+
+    pInfo->lastKey = primaryKey[index];
+    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, index));
+
+    pInfo->dOutput += ((pInfo->lastValue + pCtx->start.val) / 2) * (pInfo->lastKey - pCtx->start.key);
+
+    pInfo->hasResult = DATA_SET_FLAG;
+    pInfo->win.skey = pCtx->start.key;
+    notNullElems++;
+    i += 1;
+  } else if (pInfo->lastKey == INT64_MIN) {
+    pInfo->lastKey = primaryKey[index];
+    GET_TYPED_DATA(pInfo->lastValue, double, pCtx->inputType, GET_INPUT_CHAR_INDEX(pCtx, index));
+
+    pInfo->hasResult = DATA_SET_FLAG;
+    pInfo->win.skey = pInfo->lastKey;
+    notNullElems++;
+    i += 1;
+  }
+
+  // calculate the value of
+  switch(pCtx->inputType) {
+    case TSDB_DATA_TYPE_TINYINT: {
+      int8_t *val = (int8_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_SMALLINT: {
+      int16_t *val = (int16_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_INT: {
+      int32_t *val = (int32_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_BIGINT: {
+      int64_t *val = (int64_t*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = (double) val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_FLOAT: {
+      float *val = (float*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    case TSDB_DATA_TYPE_DOUBLE: {
+      double *val = (double*) GET_INPUT_CHAR_INDEX(pCtx, index);
+      for (; i < size; i++) {
+        if (pCtx->hasNull && isNull((const char*) &val[i], pCtx->inputType)) {
+          continue;
+        }
+
+        pInfo->dOutput += ((val[i] + pInfo->lastValue) / 2) * (primaryKey[i + index] - pInfo->lastKey);
+        pInfo->lastValue = val[i];
+        pInfo->lastKey = primaryKey[i + index];
+      }
+      break;
+    }
+    default: assert(0);
+  }
+
+  // the last interpolated time window value
+  if (pCtx->end.key != INT64_MIN) {
+    pInfo->dOutput  += ((pInfo->lastValue + pCtx->end.val) / 2) * (pCtx->end.key - pInfo->lastKey);
+    pInfo->lastValue = pCtx->end.val;
+    pInfo->lastKey   = pCtx->end.key;
+  }
+
+  pInfo->win.ekey  = pInfo->lastKey;
+
   SET_VAL(pCtx, notNullElems, 1);
 
+  if (notNullElems > 0) {
+    pResInfo->hasResult = DATA_SET_FLAG;
+  }
+
   if (pCtx->stableQuery) {
-    SResultRowCellInfo *pResInfo = GET_RES_INFO(pCtx);
     memcpy(pCtx->aOutputBuf, GET_ROWCELL_INTERBUF(pResInfo), sizeof(STwaInfo));
   }
 }

src/client/src/tscLocalMerge.c

@@ -721,10 +721,16 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
 
   // final result depends on the fields number
   memset(pSchema, 0, sizeof(SSchema) * size);
+
   for (int32_t i = 0; i < size; ++i) {
     SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
 
-    SSchema *p1 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pExpr->colInfo.colIndex);
+    SSchema p1 = {0};
+    if (pExpr->colInfo.colIndex != TSDB_TBNAME_COLUMN_INDEX) {
+      p1 = *tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pExpr->colInfo.colIndex);
+    } else {
+      p1 = tGetTableNameColumnSchema();
+    }
 
     int32_t inter = 0;
     int16_t type = -1;
@@ -743,7 +749,8 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
         functionId = TSDB_FUNC_LAST;
       }
 
-      getResultDataInfo(p1->type, p1->bytes, functionId, 0, &type, &bytes, &inter, 0, false);
+      int32_t ret = getResultDataInfo(p1.type, p1.bytes, functionId, 0, &type, &bytes, &inter, 0, false);
+      assert(ret == TSDB_CODE_SUCCESS);
     }
 
     pSchema[i].type = (uint8_t)type;

src/client/src/tscParseInsert.c

@@ -630,11 +630,17 @@ int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int3
   return TSDB_CODE_SUCCESS;
 }
 
-static void tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows) {
+static int32_t tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows) {
   pBlocks->tid = pTableMeta->id.tid;
   pBlocks->uid = pTableMeta->id.uid;
   pBlocks->sversion = pTableMeta->sversion;
-  pBlocks->numOfRows += numOfRows;
+
+  if (pBlocks->numOfRows + numOfRows >= INT16_MAX) {
+    return TSDB_CODE_TSC_INVALID_SQL;
+  } else {
+    pBlocks->numOfRows += numOfRows;
+    return TSDB_CODE_SUCCESS;
+  }
 }
 
 // data block is disordered, sort it in ascending order
@@ -722,7 +728,11 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableList, char **st
   }
 
   SSubmitBlk *pBlocks = (SSubmitBlk *)(dataBuf->pData);
-  tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
+  code = tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
+  if (code != TSDB_CODE_SUCCESS) {
+    tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", *str);
+    return code;
+  }
 
   dataBuf->vgId = pTableMeta->vgroupInfo.vgId;
   dataBuf->numOfTables = 1;
@@ -1384,7 +1394,10 @@ static int doPackSendDataBlock(SSqlObj *pSql, int32_t numOfRows, STableDataBlock
   STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0)->pTableMeta;
 
   SSubmitBlk *pBlocks = (SSubmitBlk *)(pTableDataBlocks->pData);
-  tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
+  code = tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
+  if (code != TSDB_CODE_SUCCESS) {
+    return tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", NULL);
+  }
 
   if ((code = tscMergeTableDataBlocks(pSql, pCmd->pDataBlocks)) != TSDB_CODE_SUCCESS) {
     return code;

src/client/src/tscProfile.c

@@ -39,6 +39,7 @@ void tscInitConnCb(void *param, TAOS_RES *result, int code) {
     tscSlowQueryConnInitialized = true;
     tscSaveSlowQueryFp(sql, NULL);
   }
+  taos_free_result(result);
 }
 
 void tscAddIntoSqlList(SSqlObj *pSql) {
@@ -69,6 +70,7 @@ void tscSaveSlowQueryFpCb(void *param, TAOS_RES *result, int code) {
   } else {
     tscDebug("success to save slow query, code:%d", code);
   }
+  taos_free_result(result);
 }
 
 void tscSaveSlowQueryFp(void *handle, void *tmrId) {

src/client/src/tscSQLParser.c

@@ -5103,7 +5103,7 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
   const int tokenDebugFlagEnd = 20;
   const SDNodeDynConfOption cfgOptions[] = {
       {"resetLog", 8},    {"resetQueryCache", 15},  {"balance", 7},     {"monitor", 7},
-      {"debugFlag", 9},   {"monitorDebugFlag", 16}, {"vDebugFlag", 10}, {"mDebugFlag", 10},
+      {"debugFlag", 9},   {"monDebugFlag", 12},     {"vDebugFlag", 10}, {"mDebugFlag", 10},
       {"cDebugFlag", 10}, {"httpDebugFlag", 13},    {"qDebugflag", 10}, {"sdbDebugFlag", 12},
       {"uDebugFlag", 10}, {"tsdbDebugFlag", 13},    {"sDebugflag", 10}, {"rpcDebugFlag", 12},
       {"dDebugFlag", 10}, {"mqttDebugFlag", 13},    {"wDebugFlag", 10}, {"tmrDebugFlag", 12},
@@ -5307,15 +5307,18 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn
 
     // keep original limitation value in globalLimit
     pQueryInfo->clauseLimit = pQueryInfo->limit.limit;
-    pQueryInfo->prjOffset = pQueryInfo->limit.offset;
+    pQueryInfo->prjOffset   = pQueryInfo->limit.offset;
+    pQueryInfo->tableLimit  = -1;
 
     if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
       /*
-       * the limitation/offset value should be removed during retrieve data from virtual node,
-       * since the global order are done in client side, so the limitation should also
-       * be done at the client side.
+       * the offset value should be removed during retrieve data from virtual node, since the
+       * global order are done in client side, so the offset is applied at the client side
+       * However, note that the maximum allowed number of result for each table should be less
+       * than or equal to the value of limit.
        */
       if (pQueryInfo->limit.limit > 0) {
+        pQueryInfo->tableLimit = pQueryInfo->limit.limit + pQueryInfo->limit.offset;
         pQueryInfo->limit.limit = -1;
       }
 

src/client/src/tscServer.c

@@ -684,6 +684,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   pQueryMsg->tagNameRelType = htons(pQueryInfo->tagCond.relType);
   pQueryMsg->numOfTags      = htonl(numOfTags);
   pQueryMsg->queryType      = htonl(pQueryInfo->type);
+  pQueryMsg->tableLimit     = htobe64(pQueryInfo->tableLimit);
   
   size_t numOfOutput = tscSqlExprNumOfExprs(pQueryInfo);
   pQueryMsg->numOfOutput = htons((int16_t)numOfOutput);  // this is the stage one output column number
@@ -2058,6 +2059,10 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
   if (pConnect->epSet.numOfEps > 0) {
     tscEpSetHtons(&pConnect->epSet);
     tscUpdateMgmtEpSet(&pConnect->epSet);
+
+    for (int i = 0; i < pConnect->epSet.numOfEps; ++i) {
+      tscDebug("%p epSet.fqdn[%d]: %s, pObj:%p", pSql, i, pConnect->epSet.fqdn[i], pObj);
+    }
   } 
 
   strcpy(pObj->sversion, pConnect->serverVersion);

src/client/src/tscSub.c

@@ -157,7 +157,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
 
   registerSqlObj(pSql);
 
-  code = tsParseSql(pSql, false);
+  code = tsParseSql(pSql, true);
   if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
     tsem_wait(&pSub->sem);
     code = pSql->res.code;
@@ -168,7 +168,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
     goto fail;
   }
 
-  if (pSql->cmd.command != TSDB_SQL_SELECT) {
+  if (pSql->cmd.command != TSDB_SQL_SELECT && pSql->cmd.command != TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
     line = __LINE__;
     code = TSDB_CODE_TSC_INVALID_SQL;
     goto fail;
@@ -182,7 +182,7 @@ fail:
     if (pSql->self != 0) {
       taosReleaseRef(tscObjRef, pSql->self);
     } else {
-      tscFreeSqlObj(pSql);
+    tscFreeSqlObj(pSql);
     }
 
     pSql = NULL;
@@ -401,9 +401,11 @@ TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char
     tscLoadSubscriptionProgress(pSub);
   }
 
-  if (!tscUpdateSubscription(pObj, pSub)) {
-    taos_unsubscribe(pSub, 1);
-    return NULL;
+  if (pSub->pSql->cmd.command == TSDB_SQL_SELECT) {
+    if (!tscUpdateSubscription(pObj, pSub)) {
+      taos_unsubscribe(pSub, 1);
+      return NULL;
+    }
   }
 
   pSub->interval = interval;
@@ -417,10 +419,80 @@ TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char
   return pSub;
 }
 
+SSqlObj* recreateSqlObj(SSub* pSub) {
+  SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
+  if (pSql == NULL) {
+    return NULL;
+  }
+
+  pSql->signature = pSql;
+  pSql->pTscObj = pSub->taos;
+
+  SSqlCmd* pCmd = &pSql->cmd;
+  SSqlRes* pRes = &pSql->res;
+  if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
+    tscFreeSqlObj(pSql);
+    return NULL;
+  }
+
+  pSql->param = pSub;
+  pSql->maxRetry = TSDB_MAX_REPLICA;
+  pSql->fp = asyncCallback;
+  pSql->fetchFp = asyncCallback;
+  pSql->sqlstr = strdup(pSub->pSql->sqlstr);
+  if (pSql->sqlstr == NULL) {
+    tscFreeSqlObj(pSql);
+    return NULL;
+  }
+
+  pRes->qhandle = 0;
+  pRes->numOfRows = 1;
+
+  int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
+  if (code != TSDB_CODE_SUCCESS) {
+    tscFreeSqlObj(pSql);
+    return NULL;
+  }
+
+  registerSqlObj(pSql);
+
+  code = tsParseSql(pSql, true);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    tsem_wait(&pSub->sem);
+    code = pSql->res.code;
+  }
+
+  if (code != TSDB_CODE_SUCCESS) {
+    taosReleaseRef(tscObjRef, pSql->self);
+    return NULL;
+  }
+
+  if (pSql->cmd.command != TSDB_SQL_SELECT) {
+    taosReleaseRef(tscObjRef, pSql->self);
+    return NULL;
+  }
+
+  return pSql;
+}
+
 TAOS_RES *taos_consume(TAOS_SUB *tsub) {
   SSub *pSub = (SSub *)tsub;
   if (pSub == NULL) return NULL;
 
+  if (pSub->pSql->cmd.command == TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
+    SSqlObj* pSql = recreateSqlObj(pSub);
+    if (pSql == NULL) {
+      return NULL;
+    }
+    if (pSub->pSql->self != 0) {
+      taosReleaseRef(tscObjRef, pSub->pSql->self);
+    } else {
+      tscFreeSqlObj(pSub->pSql);
+    }
+    pSub->pSql = pSql;
+    pSql->pSubscription = pSub;
+  }
+
   tscSaveSubscriptionProgress(pSub);
 
   SSqlObj *pSql = pSub->pSql;
@@ -512,10 +584,13 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
   }
 
   if (pSub->pSql != NULL) {
-    taos_free_result(pSub->pSql);
+    if (pSub->pSql->self != 0) {
+      taosReleaseRef(tscObjRef, pSub->pSql->self);
+    } else {
+      tscFreeSqlObj(pSub->pSql);
+    }
   }
 
-  tscFreeSqlObj(pSub->pSql);
   taosArrayDestroy(pSub->progress);
   tsem_destroy(&pSub->sem);
   memset(pSub, 0, sizeof(*pSub));

src/client/src/tscSystem.c

@@ -144,7 +144,7 @@ void taos_init_imp(void) {
   int64_t refreshTime = 10; // 10 seconds by default
   if (tscMetaCache == NULL) {
     tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, false, tscFreeTableMetaHelper, "tableMeta");
-    tscObjRef = taosOpenRef(4096, tscFreeRegisteredSqlObj);
+    tscObjRef = taosOpenRef(40960, tscFreeRegisteredSqlObj);
   }
 
   tscRefId = taosOpenRef(200, tscCloseTscObj);

src/common/inc/tglobal.h

@@ -56,6 +56,7 @@ extern char     tsTempDir[];
 
 //query buffer management
 extern int32_t  tsQueryBufferSize;      // maximum allowed usage buffer for each data node during query processing
+extern int32_t  tsHalfCoresForQuery;         // only 50% will be used in query processing
 
 // client
 extern int32_t tsTableMetaKeepTimer;
@@ -125,6 +126,9 @@ extern char    tsMonitorDbName[];
 extern char    tsInternalPass[];
 extern int32_t tsMonitorInterval;
 
+// stream
+extern int32_t tsEnableStream;
+
 // internal
 extern int32_t tsPrintAuth;
 extern int32_t tscEmbedded;
@@ -176,7 +180,7 @@ extern int32_t tmrDebugFlag;
 extern int32_t sdbDebugFlag;
 extern int32_t httpDebugFlag;
 extern int32_t mqttDebugFlag;
-extern int32_t monitorDebugFlag;
+extern int32_t monDebugFlag;
 extern int32_t uDebugFlag;
 extern int32_t rpcDebugFlag;
 extern int32_t odbcDebugFlag;

src/common/src/tglobal.c

@@ -107,6 +107,9 @@ int64_t tsMaxRetentWindow = 24 * 3600L;  // maximum time window tolerance
 // positive value (in MB)
 int32_t tsQueryBufferSize = -1;
 
+// only 50% cpu will be used in query processing in dnode
+int32_t tsHalfCoresForQuery = 0;
+
 // db parameters
 int32_t tsCacheBlockSize = TSDB_DEFAULT_CACHE_BLOCK_SIZE;
 int32_t tsBlocksPerVnode = TSDB_DEFAULT_TOTAL_BLOCKS;
@@ -161,6 +164,9 @@ char    tsMonitorDbName[TSDB_DB_NAME_LEN] = "log";
 char    tsInternalPass[] = "secretkey";
 int32_t tsMonitorInterval = 30;  // seconds
 
+// stream
+int32_t tsEnableStream = 1;
+
 // internal
 int32_t tsPrintAuth = 0;
 int32_t tscEmbedded = 0;
@@ -206,7 +212,7 @@ int32_t jniDebugFlag = 131;
 int32_t odbcDebugFlag = 131;
 int32_t httpDebugFlag = 131;
 int32_t mqttDebugFlag = 131;
-int32_t monitorDebugFlag = 131;
+int32_t monDebugFlag = 131;
 int32_t qDebugFlag = 131;
 int32_t rpcDebugFlag = 131;
 int32_t uDebugFlag = 131;
@@ -216,9 +222,9 @@ int32_t wDebugFlag = 135;
 int32_t tsdbDebugFlag = 131;
 int32_t cqDebugFlag = 135;
 
-int32_t (*monitorStartSystemFp)() = NULL;
-void (*monitorStopSystemFp)() = NULL;
-void (*monitorExecuteSQLFp)(char *sql) = NULL;
+int32_t (*monStartSystemFp)() = NULL;
+void (*monStopSystemFp)() = NULL;
+void (*monExecuteSQLFp)(char *sql) = NULL;
 
 char *qtypeStr[] = {"rpc", "fwd", "wal", "cq", "query"};
 
@@ -235,7 +241,7 @@ void taosSetAllDebugFlag() {
     odbcDebugFlag = debugFlag;
     httpDebugFlag = debugFlag;
     mqttDebugFlag = debugFlag;
-    monitorDebugFlag = debugFlag;
+    monDebugFlag = debugFlag;
     qDebugFlag = debugFlag;    
     rpcDebugFlag = debugFlag;
     uDebugFlag = debugFlag;
@@ -276,15 +282,15 @@ bool taosCfgDynamicOptions(char *msg) {
 
     if (strncasecmp(cfg->option, "monitor", olen) == 0) {
       if (1 == vint) {
-        if (monitorStartSystemFp) {
-          (*monitorStartSystemFp)();
+        if (monStartSystemFp) {
+          (*monStartSystemFp)();
           uInfo("monitor is enabled");
         } else {
           uError("monitor can't be updated, for monitor not initialized");
         }
       } else {
-        if (monitorStopSystemFp) {
-          (*monitorStopSystemFp)();
+        if (monStopSystemFp) {
+          (*monStopSystemFp)();
           uInfo("monitor is disabled");
         } else {
           uError("monitor can't be updated, for monitor not initialized");
@@ -307,8 +313,8 @@ bool taosCfgDynamicOptions(char *msg) {
   }
 
   if (strncasecmp(option, "resetQueryCache", 15) == 0) {
-    if (monitorExecuteSQLFp) {
-      (*monitorExecuteSQLFp)("resetQueryCache");
+    if (monExecuteSQLFp) {
+      (*monExecuteSQLFp)("resetQueryCache");
       uInfo("resetquerycache is executed");
     } else {
       uError("resetquerycache can't be executed, for monitor not started");
@@ -881,6 +887,16 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_BYTE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "halfCoresForQuery";
+  cfg.ptr = &tsHalfCoresForQuery;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = 0;
+  cfg.maxValue = 1;
+  cfg.ptrLength = 1;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   // locale & charset
   cfg.option = "timezone";
   cfg.ptr = tsTimezone;
@@ -1015,6 +1031,16 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
+  cfg.option = "stream";
+  cfg.ptr = &tsEnableStream;
+  cfg.valType = TAOS_CFG_VTYPE_INT32;
+  cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
+  cfg.minValue = 0;
+  cfg.maxValue = 1;
+  cfg.ptrLength = 1;
+  cfg.unitType = TAOS_CFG_UTYPE_NONE;
+  taosInitConfigOption(cfg);
+
   cfg.option = "httpEnableRecordSql";
   cfg.ptr = &tsHttpEnableRecordSql;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
@@ -1227,8 +1253,8 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "monitorDebugFlag";
-  cfg.ptr = &monitorDebugFlag;
+  cfg.option = "monDebugFlag";
+  cfg.ptr = &monDebugFlag;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG;
   cfg.minValue = 0;
@@ -1277,7 +1303,7 @@ static void doInitGlobalConfig(void) {
   cfg.unitType = TAOS_CFG_UTYPE_NONE;
   taosInitConfigOption(cfg);
 
-  cfg.option = "tscEnableRecordSql";
+  cfg.option = "enableRecordSql";
   cfg.ptr = &tsTscEnableRecordSql;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;

src/cq/src/cqMain.c

@@ -69,6 +69,9 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row);
 static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
 
 void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
+  if (tsEnableStream == 0) {
+    return NULL;
+  }
   SCqContext *pContext = calloc(sizeof(SCqContext), 1);
   if (pContext == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
@@ -99,6 +102,9 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
 }
 
 void cqClose(void *handle) {
+  if (tsEnableStream == 0) {
+    return;
+  }
   SCqContext *pContext = handle;
   if (handle == NULL) return;
 
@@ -129,6 +135,9 @@ void cqClose(void *handle) {
 }
 
 void cqStart(void *handle) {
+  if (tsEnableStream == 0) {
+    return;
+  }
   SCqContext *pContext = handle;
   if (pContext->dbConn || pContext->master) return;
 
@@ -147,6 +156,9 @@ void cqStart(void *handle) {
 }
 
 void cqStop(void *handle) {
+  if (tsEnableStream == 0) {
+    return;
+  }
   SCqContext *pContext = handle;
   cInfo("vgId:%d, stop all CQs", pContext->vgId);
   if (pContext->dbConn == NULL || pContext->master == 0) return;
@@ -174,6 +186,9 @@ void cqStop(void *handle) {
 }
 
 void *cqCreate(void *handle, uint64_t uid, int32_t tid, char *sqlStr, STSchema *pSchema) {
+  if (tsEnableStream == 0) {
+    return NULL;
+  }
   SCqContext *pContext = handle;
 
   SCqObj *pObj = calloc(sizeof(SCqObj), 1);
@@ -203,6 +218,9 @@ void *cqCreate(void *handle, uint64_t uid, int32_t tid, char *sqlStr, STSchema *
 }
 
 void cqDrop(void *handle) {
+  if (tsEnableStream == 0) {
+    return;
+  }
   SCqObj *pObj = handle;
   SCqContext *pContext = pObj->pContext;
 

src/dnode/src/dnodeMgmt.c

@@ -24,7 +24,7 @@
 #include "tqueue.h"
 #include "tsync.h"
 #include "ttimer.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tglobal.h"
 #include "dnode.h"
 #include "vnode.h"

src/dnode/src/dnodeModule.c

@@ -78,10 +78,10 @@ static void dnodeAllocModules() {
 
   tsModule[TSDB_MOD_MONITOR].enable    = (tsEnableMonitorModule == 1);
   tsModule[TSDB_MOD_MONITOR].name      = "monitor";
-  tsModule[TSDB_MOD_MONITOR].initFp    = monitorInitSystem;
-  tsModule[TSDB_MOD_MONITOR].cleanUpFp = monitorCleanUpSystem;
-  tsModule[TSDB_MOD_MONITOR].startFp   = monitorStartSystem;
-  tsModule[TSDB_MOD_MONITOR].stopFp    = monitorStopSystem;
+  tsModule[TSDB_MOD_MONITOR].initFp    = monInitSystem;
+  tsModule[TSDB_MOD_MONITOR].cleanUpFp = monCleanupSystem;
+  tsModule[TSDB_MOD_MONITOR].startFp   = monStartSystem;
+  tsModule[TSDB_MOD_MONITOR].stopFp    = monStopSystem;
   if (tsEnableMonitorModule) {
     dnodeSetModuleStatus(TSDB_MOD_MONITOR);
   }

src/dnode/src/dnodePeer.c

@@ -182,6 +182,8 @@ void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) {
 void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
   SRpcEpSet epSet = {0};
   dnodeGetEpSetForPeer(&epSet);
+
+  assert(tsClientRpc != 0);
   rpcSendRecv(tsClientRpc, &epSet, rpcMsg, rpcRsp);
 }
 

src/inc/monitor.h

@@ -47,13 +47,13 @@ typedef struct {
   int8_t  accessState;
 } SAcctMonitorObj;
 
-int32_t monitorInitSystem();
-int32_t monitorStartSystem();
-void    monitorStopSystem();
-void    monitorCleanUpSystem();
-void    monitorSaveAcctLog(SAcctMonitorObj *pMonObj);
-void    monitorSaveLog(int32_t level, const char *const format, ...);
-void    monitorExecuteSQL(char *sql);
+int32_t monInitSystem();
+int32_t monStartSystem();
+void    monStopSystem();
+void    monCleanupSystem();
+void    monSaveAcctLog(SAcctMonitorObj *pMonObj);
+void    monSaveLog(int32_t level, const char *const format, ...);
+void    monExecuteSQL(char *sql);
 
 #ifdef __cplusplus
 }

src/inc/taosmsg.h

@@ -476,19 +476,21 @@ typedef struct {
   int16_t     numOfGroupCols;   // num of group by columns
   int16_t     orderByIdx;
   int16_t     orderType;        // used in group by xx order by xxx
+  int64_t     tableLimit;       // limit the number of rows for each table, used in order by + limit in stable projection query.
+  int16_t     prjOrder;         // global order in super table projection query.
   int64_t     limit;
   int64_t     offset;
   uint32_t    queryType;        // denote another query process
   int16_t     numOfOutput;      // final output columns numbers
   int16_t     tagNameRelType;   // relation of tag criteria and tbname criteria
-  int16_t     fillType;      // interpolate type
-  uint64_t    fillVal;       // default value array list
+  int16_t     fillType;         // interpolate type
+  uint64_t    fillVal;          // default value array list
   int32_t     secondStageOutput;
-  int32_t     tsOffset;       // offset value in current msg body, NOTE: ts list is compressed
-  int32_t     tsLen;          // total length of ts comp block
-  int32_t     tsNumOfBlocks;  // ts comp block numbers
-  int32_t     tsOrder;        // ts comp block order
-  int32_t     numOfTags;      // number of tags columns involved
+  int32_t     tsOffset;         // offset value in current msg body, NOTE: ts list is compressed
+  int32_t     tsLen;            // total length of ts comp block
+  int32_t     tsNumOfBlocks;    // ts comp block numbers
+  int32_t     tsOrder;          // ts comp block order
+  int32_t     numOfTags;        // number of tags columns involved
   SColumnInfo colList[];
 } SQueryTableMsg;
 

src/inc/tbalance.h → src/inc/tbn.h

@@ -23,14 +23,14 @@ extern "C" {
 struct SVgObj;
 struct SDnodeObj;
 
-int32_t balanceInit();
-void    balanceCleanUp();
-void    balanceAsyncNotify();
-void    balanceSyncNotify();
-void    balanceReset();
-int32_t balanceAllocVnodes(struct SVgObj *pVgroup);
-int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId);
-int32_t balanceDropDnode(struct SDnodeObj *pDnode);
+int32_t bnInit();
+void    bnCleanUp();
+void    bnNotify();
+void    bnCheckModules();
+void    bnReset();
+int32_t bnAllocVnodes(struct SVgObj *pVgroup);
+int32_t bnAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId);
+int32_t bnDropDnode(struct SDnodeObj *pDnode);
 
 #ifdef __cplusplus
 }

src/mnode/inc/mnodeInt.h

@@ -41,9 +41,9 @@ extern int32_t sdbDebugFlag;
 #define sdbDebug(...) { if (sdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
 #define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
 
-#define mLError(...) { monitorSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) }
-#define mLWarn(...)  { monitorSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)  }
-#define mLInfo(...)  { monitorSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
+#define mLError(...) { monSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) }
+#define mLWarn(...)  { monSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)  }
+#define mLInfo(...)  { monSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
 
 #ifdef __cplusplus
 }

src/mnode/src/mnodeDb.c

@@ -20,7 +20,7 @@
 #include "tgrant.h"
 #include "tglobal.h"
 #include "tname.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tdataformat.h"
 #include "mnode.h"
 #include "mnodeDef.h"
@@ -250,7 +250,7 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
   }
 
   if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > pCfg->daysToKeep) {
-    mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep, TSDB_MIN_KEEP, pCfg->daysToKeep);
+    mError("invalid db option daysToKeep2:%d valid range: [%d, %d]", pCfg->daysToKeep2, TSDB_MIN_KEEP, pCfg->daysToKeep);
     return TSDB_CODE_MND_INVALID_DB_OPTION_KEEP;
   }
 
@@ -1004,7 +1004,7 @@ static int32_t mnodeAlterDbCb(SMnodeMsg *pMsg, int32_t code) {
   mDebug("db:%s, all vgroups is altered", pDb->name);
   mLInfo("db:%s, is alterd by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
 
-  balanceAsyncNotify();
+  bnNotify();
 
   return TSDB_CODE_SUCCESS;
 }

src/mnode/src/mnodeDnode.c

@@ -16,12 +16,12 @@
 #define _DEFAULT_SOURCE
 #include "os.h"
 #include "tgrant.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tglobal.h"
 #include "tconfig.h"
 #include "tutil.h"
 #include "tsocket.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tsync.h"
 #include "tdataformat.h"
 #include "mnode.h"
@@ -115,7 +115,7 @@ static int32_t mnodeDnodeActionDelete(SSdbRow *pRow) {
   mnodeDropAllDnodeVgroups(pDnode);
 #endif  
   mnodeDropMnodeLocal(pDnode->dnodeId);
-  balanceAsyncNotify();
+  bnNotify();
   mnodeUpdateDnodeEps();
 
   mDebug("dnode:%d, all vgroups is dropped from sdb", pDnode->dnodeId);
@@ -347,7 +347,7 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
       return TSDB_CODE_MND_INVALID_DNODE_CFG_OPTION;
     }
 
-    int32_t code = balanceAlterDnode(pDnode, vnodeId, dnodeId);
+    int32_t code = bnAlterDnode(pDnode, vnodeId, dnodeId);
     mnodeDecDnodeRef(pDnode);
     return code;
   } else {
@@ -591,8 +591,8 @@ static int32_t mnodeProcessDnodeStatusMsg(SMnodeMsg *pMsg) {
     mInfo("dnode:%d, from offline to online", pDnode->dnodeId);
     pDnode->status = TAOS_DN_STATUS_READY;
     pDnode->offlineReason = TAOS_DN_OFF_ONLINE;
-    balanceSyncNotify();
-    balanceAsyncNotify();
+    bnCheckModules();
+    bnNotify();
   }
 
   if (openVnodes != pDnode->openVnodes) {
@@ -708,7 +708,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
 #ifndef _SYNC
   int32_t code = mnodeDropDnode(pDnode, pMsg);
 #else
-  int32_t code = balanceDropDnode(pDnode);
+  int32_t code = bnDropDnode(pDnode);
 #endif  
   mnodeDecDnodeRef(pDnode);
   return code;
@@ -1182,12 +1182,12 @@ static char* mnodeGetDnodeAlternativeRoleStr(int32_t alternativeRole) {
 
 #ifndef _SYNC
 
-int32_t balanceInit() { return TSDB_CODE_SUCCESS; }
-void    balanceCleanUp() {}
-void    balanceAsyncNotify() {}
-void    balanceSyncNotify() {}
-void    balanceReset() {}
-int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; }
+int32_t bnInit() { return TSDB_CODE_SUCCESS; }
+void    bnCleanUp() {}
+void    bnNotify() {}
+void    bnCheckModules() {}
+void    bnReset() {}
+int32_t bnAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; }
 
 char* syncRole[] = {
   "offline",
@@ -1197,7 +1197,7 @@ char* syncRole[] = {
   "master"
 };
 
-int32_t balanceAllocVnodes(SVgObj *pVgroup) {
+int32_t bnAllocVnodes(SVgObj *pVgroup) {
   void *     pIter = NULL;
   SDnodeObj *pDnode = NULL;
   SDnodeObj *pSelDnode = NULL;

src/mnode/src/mnodeMain.c

@@ -17,7 +17,7 @@
 #include "os.h"
 #include "taosdef.h"
 #include "tsched.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tgrant.h"
 #include "ttimer.h"
 #include "tglobal.h"
@@ -58,7 +58,7 @@ static const SMnodeComponent tsMnodeComponents[] = {
   {"tables",  mnodeInitTables,  mnodeCleanupTables},  
   {"mnodes",  mnodeInitMnodes,  mnodeCleanupMnodes},
   {"sdb",     sdbInit,          sdbCleanUp},
-  {"balance", balanceInit,      balanceCleanUp},
+  {"balance", bnInit,           bnCleanUp},
   {"grant",   grantInit,        grantCleanUp},
   {"show",    mnodeInitShow,    mnodeCleanUpShow}
 };

src/mnode/src/mnodeMnode.c

@@ -19,7 +19,7 @@
 #include "tglobal.h"
 #include "trpc.h"
 #include "tsync.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tutil.h"
 #include "tsocket.h"
 #include "tdataformat.h"

src/mnode/src/mnodePeer.c

@@ -20,7 +20,7 @@
 #include "tsystem.h"
 #include "tutil.h"
 #include "tgrant.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tglobal.h"
 #include "mnode.h"
 #include "dnode.h"

src/mnode/src/mnodeRead.c

@@ -17,7 +17,7 @@
 #include "os.h"
 #include "taosdef.h"
 #include "tsched.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tgrant.h"
 #include "ttimer.h"
 #include "tglobal.h"

src/mnode/src/mnodeSdb.c

@@ -19,7 +19,7 @@
 #include "hash.h"
 #include "tutil.h"
 #include "tref.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tqueue.h"
 #include "twal.h"
 #include "tsync.h"
@@ -244,7 +244,7 @@ static void sdbNotifyRole(int32_t vgId, int8_t role) {
   sdbInfo("vgId:1, mnode role changed from %s to %s", syncRole[tsSdbMgmt.role], syncRole[role]);
 
   if (role == TAOS_SYNC_ROLE_MASTER && tsSdbMgmt.role != TAOS_SYNC_ROLE_MASTER) {
-    balanceReset();
+    bnReset();
   }
   tsSdbMgmt.role = role;
 

src/mnode/src/mnodeVgroup.c

@@ -20,7 +20,7 @@
 #include "tsocket.h"
 #include "tidpool.h"
 #include "tsync.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tglobal.h"
 #include "tdataformat.h"
 #include "dnode.h"
@@ -563,7 +563,7 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg) {
   pVgroup->numOfVnodes = pDb->cfg.replications;
   pVgroup->createdTime = taosGetTimestampMs();
   pVgroup->accessState = TSDB_VN_ALL_ACCCESS;
-  int32_t code = balanceAllocVnodes(pVgroup);
+  int32_t code = bnAllocVnodes(pVgroup);
   if (code != TSDB_CODE_SUCCESS) {
     mError("db:%s, no enough dnode to alloc %d vnodes to vgroup, reason:%s", pDb->name, pVgroup->numOfVnodes,
            tstrerror(code));

src/mnode/src/mnodeWrite.c

@@ -17,7 +17,7 @@
 #include "os.h"
 #include "taosdef.h"
 #include "tsched.h"
-#include "tbalance.h"
+#include "tbn.h"
 #include "tgrant.h"
 #include "tglobal.h"
 #include "trpc.h"

src/plugins/monitor/src/monitorMain.c → src/plugins/monitor/src/monMain.c

@@ -27,12 +27,12 @@
 #include "monitor.h"
 #include "taoserror.h"
 
-#define mnFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
-#define mnError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
-#define mnWarn(...)  { if (monitorDebugFlag & DEBUG_WARN)  { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }}
-#define mnInfo(...)  { if (monitorDebugFlag & DEBUG_INFO)  { taosPrintLog("MON ", 255, __VA_ARGS__); }}
-#define mnDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
-#define mnTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
+#define monFatal(...) { if (monDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
+#define monError(...) { if (monDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
+#define monWarn(...)  { if (monDebugFlag & DEBUG_WARN)  { taosPrintLog("MON WARN ", 255, __VA_ARGS__); }}
+#define monInfo(...)  { if (monDebugFlag & DEBUG_INFO)  { taosPrintLog("MON ", 255, __VA_ARGS__); }}
+#define monDebug(...) { if (monDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monDebugFlag, __VA_ARGS__); }}
+#define monTrace(...) { if (monDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monDebugFlag, __VA_ARGS__); }}
 
 #define SQL_LENGTH     1030
 #define LOG_LEN_STR    100
@@ -48,12 +48,12 @@ typedef enum {
   MON_CMD_CREATE_TB_ACCT_ROOT,
   MON_CMD_CREATE_TB_SLOWQUERY,
   MON_CMD_MAX
-} EMonitorCommand;
+} EMonCmd;
 
 typedef enum {
   MON_STATE_NOT_INIT,
   MON_STATE_INITED
-} EMonitorState;
+} EMonState;
 
 typedef struct {
   pthread_t thread;
@@ -64,17 +64,17 @@ typedef struct {
   int8_t    start;   // enable/disable by mnode
   int8_t    quiting; // taosd is quiting 
   char      sql[SQL_LENGTH + 1];
-} SMonitorConn;
+} SMonConn;
 
-static SMonitorConn tsMonitor = {0};
-static void  monitorSaveSystemInfo();
-static void *monitorThreadFunc(void *param);
-static void  monitorBuildMonitorSql(char *sql, int32_t cmd);
-extern int32_t (*monitorStartSystemFp)();
-extern void    (*monitorStopSystemFp)();
-extern void    (*monitorExecuteSQLFp)(char *sql);
+static SMonConn tsMonitor = {0};
+static void  monSaveSystemInfo();
+static void *monThreadFunc(void *param);
+static void  monBuildMonitorSql(char *sql, int32_t cmd);
+extern int32_t (*monStartSystemFp)();
+extern void    (*monStopSystemFp)();
+extern void    (*monExecuteSQLFp)(char *sql);
 
-int32_t monitorInitSystem() {
+int32_t monInitSystem() {
   if (tsMonitor.ep[0] == 0) {
     strcpy(tsMonitor.ep, tsLocalEp);
   }
@@ -90,29 +90,29 @@ int32_t monitorInitSystem() {
   pthread_attr_init(&thAttr);
   pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
 
-  if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) {
-    mnError("failed to create thread to for monitor module, reason:%s", strerror(errno));
+  if (pthread_create(&tsMonitor.thread, &thAttr, monThreadFunc, NULL)) {
+    monError("failed to create thread to for monitor module, reason:%s", strerror(errno));
     return -1;
   }
 
   pthread_attr_destroy(&thAttr);
-  mnDebug("monitor thread is launched");
+  monDebug("monitor thread is launched");
 
-  monitorStartSystemFp = monitorStartSystem;
-  monitorStopSystemFp = monitorStopSystem;
+  monStartSystemFp = monStartSystem;
+  monStopSystemFp = monStopSystem;
   return 0;
 }
 
-int32_t monitorStartSystem() {
+int32_t monStartSystem() {
   taos_init();
   tsMonitor.start = 1;
-  monitorExecuteSQLFp = monitorExecuteSQL;
-  mnInfo("monitor module start");
+  monExecuteSQLFp = monExecuteSQL;
+  monInfo("monitor module start");
   return 0;
 }
 
-static void *monitorThreadFunc(void *param) {
-  mnDebug("starting to initialize monitor module ...");
+static void *monThreadFunc(void *param) {
+  monDebug("starting to initialize monitor module ...");
 
   while (1) {
     static int32_t accessTimes = 0;
@@ -121,7 +121,7 @@ static void *monitorThreadFunc(void *param) {
 
     if (tsMonitor.quiting) {
       tsMonitor.state = MON_STATE_NOT_INIT;
-      mnInfo("monitor thread will quit, for taosd is quiting");
+      monInfo("monitor thread will quit, for taosd is quiting");
       break;
     } else {
       taosGetDisk();
@@ -132,7 +132,7 @@ static void *monitorThreadFunc(void *param) {
     }
     
     if (dnodeGetDnodeId() <= 0) {
-      mnDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
+      monDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
       continue;
     }
 
@@ -140,10 +140,10 @@ static void *monitorThreadFunc(void *param) {
       tsMonitor.state = MON_STATE_NOT_INIT;
       tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0);
       if (tsMonitor.conn == NULL) {
-        mnError("failed to connect to database, reason:%s", tstrerror(terrno));
+        monError("failed to connect to database, reason:%s", tstrerror(terrno));
         continue;
       } else {
-        mnDebug("connect to database success");
+        monDebug("connect to database success");
       }
     }
 
@@ -151,16 +151,16 @@ static void *monitorThreadFunc(void *param) {
       int code = 0;
 
       for (; tsMonitor.cmdIndex < MON_CMD_MAX; ++tsMonitor.cmdIndex) {
-        monitorBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex);
+        monBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex);
         void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
         code = taos_errno(res);
         taos_free_result(res);
 
         if (code != 0) {
-          mnError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
+          monError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
           break;
         } else {
-          mnDebug("successfully to exec sql:%s", tsMonitor.sql);
+          monDebug("successfully to exec sql:%s", tsMonitor.sql);
         }
       }
 
@@ -171,16 +171,16 @@ static void *monitorThreadFunc(void *param) {
 
     if (tsMonitor.state == MON_STATE_INITED) {
       if (accessTimes % tsMonitorInterval == 0) {
-        monitorSaveSystemInfo();
+        monSaveSystemInfo();
       }
     }
   }
 
-  mnInfo("monitor thread is stopped");
+  monInfo("monitor thread is stopped");
   return NULL;
 }
 
-static void monitorBuildMonitorSql(char *sql, int32_t cmd) {
+static void monBuildMonitorSql(char *sql, int32_t cmd) {
   memset(sql, 0, SQL_LENGTH);
 
   if (cmd == MON_CMD_CREATE_DB) {
@@ -236,47 +236,47 @@ static void monitorBuildMonitorSql(char *sql, int32_t cmd) {
   sql[SQL_LENGTH] = 0;
 }
 
-void monitorStopSystem() {
+void monStopSystem() {
   tsMonitor.start = 0;
   tsMonitor.state = MON_STATE_NOT_INIT;
-  monitorExecuteSQLFp = NULL;
-  mnInfo("monitor module stopped");
+  monExecuteSQLFp = NULL;
+  monInfo("monitor module stopped");
 }
 
-void monitorCleanUpSystem() {
+void monCleanupSystem() {
   tsMonitor.quiting = 1;
-  monitorStopSystem();
+  monStopSystem();
   pthread_join(tsMonitor.thread, NULL);
   if (tsMonitor.conn != NULL) {
     taos_close(tsMonitor.conn);
     tsMonitor.conn = NULL;
   }
-  mnInfo("monitor module is cleaned up");
+  monInfo("monitor module is cleaned up");
 }
 
 // unit is MB
-static int32_t monitorBuildMemorySql(char *sql) {
+static int32_t monBuildMemorySql(char *sql) {
   float sysMemoryUsedMB = 0;
   bool  suc = taosGetSysMemory(&sysMemoryUsedMB);
   if (!suc) {
-    mnDebug("failed to get sys memory info");
+    monDebug("failed to get sys memory info");
   }
 
   float procMemoryUsedMB = 0;
   suc = taosGetProcMemory(&procMemoryUsedMB);
   if (!suc) {
-    mnDebug("failed to get proc memory info");
+    monDebug("failed to get proc memory info");
   }
 
   return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
 }
 
 // unit is %
-static int32_t monitorBuildCpuSql(char *sql) {
+static int32_t monBuildCpuSql(char *sql) {
   float sysCpuUsage = 0, procCpuUsage = 0;
   bool  suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
   if (!suc) {
-    mnDebug("failed to get cpu usage");
+    monDebug("failed to get cpu usage");
   }
 
   if (sysCpuUsage <= procCpuUsage) {
@@ -287,72 +287,72 @@ static int32_t monitorBuildCpuSql(char *sql) {
 }
 
 // unit is GB
-static int32_t monitorBuildDiskSql(char *sql) {
+static int32_t monBuildDiskSql(char *sql) {
   return sprintf(sql, ", %f, %d", (tsTotalDataDirGB - tsAvailDataDirGB), (int32_t)tsTotalDataDirGB);
 }
 
 // unit is Kb
-static int32_t monitorBuildBandSql(char *sql) {
+static int32_t monBuildBandSql(char *sql) {
   float bandSpeedKb = 0;
   bool  suc = taosGetBandSpeed(&bandSpeedKb);
   if (!suc) {
-    mnDebug("failed to get bandwidth speed");
+    monDebug("failed to get bandwidth speed");
   }
 
   return sprintf(sql, ", %f", bandSpeedKb);
 }
 
-static int32_t monitorBuildReqSql(char *sql) {
+static int32_t monBuildReqSql(char *sql) {
   SStatisInfo info = dnodeGetStatisInfo();
   return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
 }
 
-static int32_t monitorBuildIoSql(char *sql) {
+static int32_t monBuildIoSql(char *sql) {
   float readKB = 0, writeKB = 0;
   bool  suc = taosGetProcIO(&readKB, &writeKB);
   if (!suc) {
-    mnDebug("failed to get io info");
+    monDebug("failed to get io info");
   }
 
   return sprintf(sql, ", %f, %f", readKB, writeKB);
 }
 
-static void monitorSaveSystemInfo() {
+static void monSaveSystemInfo() {
   int64_t ts = taosGetTimestampUs();
   char *  sql = tsMonitor.sql;
   int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn%d values(%" PRId64, tsMonitorDbName, dnodeGetDnodeId(), ts);
 
-  pos += monitorBuildCpuSql(sql + pos);
-  pos += monitorBuildMemorySql(sql + pos);
-  pos += monitorBuildDiskSql(sql + pos);
-  pos += monitorBuildBandSql(sql + pos);
-  pos += monitorBuildIoSql(sql + pos);
-  pos += monitorBuildReqSql(sql + pos);
+  pos += monBuildCpuSql(sql + pos);
+  pos += monBuildMemorySql(sql + pos);
+  pos += monBuildDiskSql(sql + pos);
+  pos += monBuildBandSql(sql + pos);
+  pos += monBuildIoSql(sql + pos);
+  pos += monBuildReqSql(sql + pos);
 
   void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
   int   code = taos_errno(res);
   taos_free_result(res);
 
   if (code != 0) {
-    mnError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
+    monError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
   } else {
-    mnDebug("successfully to save system info, sql:%s", tsMonitor.sql);
+    monDebug("successfully to save system info, sql:%s", tsMonitor.sql);
   }
 }
 
-static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
+static void monExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
   int32_t c = taos_errno(result);
   if (c != TSDB_CODE_SUCCESS) {
-    mnError("save %s failed, reason:%s", (char *)param, tstrerror(c));
+    monError("save %s failed, reason:%s", (char *)param, tstrerror(c));
   } else {
     int32_t rows = taos_affected_rows(result);
-    mnDebug("save %s succ, rows:%d", (char *)param, rows);
+    monDebug("save %s succ, rows:%d", (char *)param, rows);
   }
 
   taos_free_result(result);
 }
 
-void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
+void monSaveAcctLog(SAcctMonitorObj *pMon) {
   if (tsMonitor.state != MON_STATE_INITED) return;
 
   char sql[1024] = {0};
@@ -382,11 +382,11 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
           pMon->totalConns, pMon->maxConns,
           pMon->accessState);
 
-  mnDebug("save account info, sql:%s", sql);
-  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info");
+  monDebug("save account info, sql:%s", sql);
+  taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "account info");
 }
 
-void monitorSaveLog(int32_t level, const char *const format, ...) {
+void monSaveLog(int32_t level, const char *const format, ...) {
   if (tsMonitor.state != MON_STATE_INITED) return;
 
   va_list argpointer;
@@ -403,13 +403,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
   len += sprintf(sql + len, "', '%s')", tsLocalEp);
   sql[len++] = 0;
 
-  mnDebug("save log, sql: %s", sql);
-  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log");
+  monDebug("save log, sql: %s", sql);
+  taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "log");
 }
 
-void monitorExecuteSQL(char *sql) {
+void monExecuteSQL(char *sql) {
   if (tsMonitor.state != MON_STATE_INITED) return;
 
-  mnDebug("execute sql:%s", sql);
-  taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql");
+  monDebug("execute sql:%s", sql);
+  taos_query_a(tsMonitor.conn, sql, monExecSqlCb, "sql");
 }

src/query/inc/qExecutor.h

@@ -83,16 +83,15 @@ typedef struct SResultRec {
   int32_t threshold;  // result size threshold in rows.
 } SResultRec;
 
-typedef struct SWindowResInfo {
-  SResultRow**   pResult;    // result list
-  int16_t        type:8;     // data type for hash key
-  int32_t        size:24;    // number of result set
-  int32_t        threshold;  // threshold to halt query and return the generated results.
-  int32_t        capacity;   // max capacity
-  int32_t        curIndex;   // current start active index
-  int64_t        startTime;  // start time of the first time window for sliding query
-  int64_t        prevSKey;   // previous (not completed) sliding window start key
-} SWindowResInfo;
+typedef struct SResultRowInfo {
+  SResultRow** pResult;    // result list
+  int16_t      type:8;     // data type for hash key
+  int32_t      size:24;    // number of result set
+  int32_t      capacity;   // max capacity
+  int32_t      curIndex;   // current start active index
+  int64_t      startTime;  // start time of the first time window for sliding query
+  int64_t      prevSKey;   // previous (not completed) sliding window start key
+} SResultRowInfo;
 
 typedef struct SColumnFilterElem {
   int16_t           bytes;  // column length
@@ -115,7 +114,7 @@ typedef struct STableQueryInfo {
   STimeWindow win;
   STSCursor   cur;
   void*       pTable;         // for retrieve the page id list
-  SWindowResInfo windowResInfo;
+  SResultRowInfo windowResInfo;
 } STableQueryInfo;
 
 typedef struct SQueryCostInfo {
@@ -179,7 +178,7 @@ typedef struct SQueryRuntimeEnv {
   uint16_t*            offset;
   uint16_t             scanFlag;         // denotes reversed scan of data or not
   SFillInfo*           pFillInfo;
-  SWindowResInfo       windowResInfo;
+  SResultRowInfo       windowResInfo;
   STSBuf*              pTSBuf;
   STSCursor            cur;
   SQueryCostInfo       summary;
@@ -190,6 +189,7 @@ typedef struct SQueryRuntimeEnv {
   bool                 groupbyNormalCol; // denote if this is a groupby normal column query
   bool                 hasTagResults;    // if there are tag values in final result or not
   bool                 timeWindowInterpo;// if the time window start/end required interpolation
+  bool                 queryWindowIdentical; // all query time windows are identical for all tables in one group
   int32_t              interBufSize;     // intermediate buffer sizse
   int32_t              prevGroupId;      // previous executed group id
   SDiskbasedResultBuf* pResultBuf;       // query result buffer based on blocked-wised disk file
@@ -217,7 +217,8 @@ typedef struct SQInfo {
   STableGroupInfo  tableGroupInfo;       // table <tid, last_key> list  SArray<STableKeyInfo>
   STableGroupInfo  tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
   SQueryRuntimeEnv runtimeEnv;
-  SArray*          arrTableIdInfo;
+//  SArray*          arrTableIdInfo;
+  SHashObj*        arrTableIdInfo;
   int32_t          groupIndex;
 
   /*

src/query/inc/qUtil.h

@@ -30,19 +30,19 @@ void clearResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pRow, int16_t typ
 void copyResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* dst, const SResultRow* src, int16_t type);
 SResultRowCellInfo* getResultCell(SQueryRuntimeEnv* pRuntimeEnv, const SResultRow* pRow, int32_t index);
 
-int32_t initWindowResInfo(SWindowResInfo* pWindowResInfo, int32_t size, int32_t threshold, int16_t type);
+int32_t initWindowResInfo(SResultRowInfo* pWindowResInfo, int32_t size, int16_t type);
 
-void    cleanupTimeWindowInfo(SWindowResInfo* pWindowResInfo);
-void    resetTimeWindowInfo(SQueryRuntimeEnv* pRuntimeEnv, SWindowResInfo* pWindowResInfo);
+void    cleanupTimeWindowInfo(SResultRowInfo* pWindowResInfo);
+void    resetTimeWindowInfo(SQueryRuntimeEnv* pRuntimeEnv, SResultRowInfo* pWindowResInfo);
 void    clearFirstNWindowRes(SQueryRuntimeEnv *pRuntimeEnv, int32_t num);
 
 void    clearClosedTimeWindow(SQueryRuntimeEnv* pRuntimeEnv);
-int32_t numOfClosedTimeWindow(SWindowResInfo* pWindowResInfo);
-void    closeTimeWindow(SWindowResInfo* pWindowResInfo, int32_t slot);
-void    closeAllTimeWindow(SWindowResInfo* pWindowResInfo);
-void    removeRedundantWindow(SWindowResInfo *pWindowResInfo, TSKEY lastKey, int32_t order);
+int32_t numOfClosedTimeWindow(SResultRowInfo* pWindowResInfo);
+void    closeTimeWindow(SResultRowInfo* pWindowResInfo, int32_t slot);
+void    closeAllTimeWindow(SResultRowInfo* pWindowResInfo);
+void    removeRedundantWindow(SResultRowInfo *pWindowResInfo, TSKEY lastKey, int32_t order);
 
-static FORCE_INLINE SResultRow *getResultRow(SWindowResInfo *pWindowResInfo, int32_t slot) {
+static FORCE_INLINE SResultRow *getResultRow(SResultRowInfo *pWindowResInfo, int32_t slot) {
   assert(pWindowResInfo != NULL && slot >= 0 && slot < pWindowResInfo->size);
   return pWindowResInfo->pResult[slot];
 }
@@ -50,7 +50,7 @@ static FORCE_INLINE SResultRow *getResultRow(SWindowResInfo *pWindowResInfo, int
 #define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
 #define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!sq))? (_q)->pExpr1[1].base.arg->argValue.i64:1)
 
-bool isWindowResClosed(SWindowResInfo *pWindowResInfo, int32_t slot);
+bool isWindowResClosed(SResultRowInfo *pWindowResInfo, int32_t slot);
 
 int32_t initResultRow(SResultRow *pResultRow);
 

src/query/src/qParserImpl.c

@@ -405,14 +405,29 @@ void tSQLSetColumnType(TAOS_FIELD *pField, SStrToken *type) {
         if (type->type == 0) {
           pField->bytes = 0;
         } else {
-          pField->bytes = (int16_t)(-(int32_t)type->type * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
+          int32_t bytes = -(int32_t)(type->type);
+          if (bytes > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
+            // we have to postpone reporting the error because it cannot be done here
+            // as pField->bytes is int16_t, use 'TSDB_MAX_NCHAR_LEN + 1' to avoid overflow
+            bytes = TSDB_MAX_NCHAR_LEN + 1;
+          } else {
+            bytes = bytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
+          }
+          pField->bytes = (int16_t)bytes;
         }
       } else if (i == TSDB_DATA_TYPE_BINARY) {
         /* for binary, the TOKENTYPE is the length of binary */
         if (type->type == 0) {
           pField->bytes = 0;
         } else {
-          pField->bytes = (int16_t) (-(int32_t) type->type + VARSTR_HEADER_SIZE);
+          int32_t bytes = -(int32_t)(type->type);
+          if (bytes > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
+            // refer comment for NCHAR above
+            bytes = TSDB_MAX_BINARY_LEN + 1;
+          } else {
+            bytes += VARSTR_HEADER_SIZE;
+          }
+          pField->bytes = (int16_t)bytes;
         } 
       }
       break;

src/query/src/qUtil.c

@@ -43,51 +43,48 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
   return size;
 }
 
-int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, int32_t size, int32_t threshold, int16_t type) {
-  pWindowResInfo->capacity = size;
-  pWindowResInfo->threshold = threshold;
-  
-  pWindowResInfo->type = type;
-  pWindowResInfo->curIndex = -1;
-  pWindowResInfo->size     = 0;
-  pWindowResInfo->prevSKey = TSKEY_INITIAL_VAL;
+int32_t initWindowResInfo(SResultRowInfo *pResultRowInfo, int32_t size, int16_t type) {
+  pResultRowInfo->capacity = size;
+
+  pResultRowInfo->type = type;
+  pResultRowInfo->curIndex = -1;
+  pResultRowInfo->size     = 0;
+  pResultRowInfo->prevSKey = TSKEY_INITIAL_VAL;
 
-  pWindowResInfo->pResult = calloc(pWindowResInfo->capacity, POINTER_BYTES);
-  if (pWindowResInfo->pResult == NULL) {
+  pResultRowInfo->pResult = calloc(pResultRowInfo->capacity, POINTER_BYTES);
+  if (pResultRowInfo->pResult == NULL) {
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
   return TSDB_CODE_SUCCESS;
 }
 
-void cleanupTimeWindowInfo(SWindowResInfo *pWindowResInfo) {
-  if (pWindowResInfo == NULL) {
+void cleanupTimeWindowInfo(SResultRowInfo *pResultRowInfo) {
+  if (pResultRowInfo == NULL) {
     return;
   }
-  if (pWindowResInfo->capacity == 0) {
-    assert(pWindowResInfo->pResult == NULL);
+  if (pResultRowInfo->capacity == 0) {
+    assert(pResultRowInfo->pResult == NULL);
     return;
   }
 
-  if (pWindowResInfo->type == TSDB_DATA_TYPE_BINARY || pWindowResInfo->type == TSDB_DATA_TYPE_NCHAR) {
-    for(int32_t i = 0; i < pWindowResInfo->size; ++i) {
-      tfree(pWindowResInfo->pResult[i]->key);
+  if (pResultRowInfo->type == TSDB_DATA_TYPE_BINARY || pResultRowInfo->type == TSDB_DATA_TYPE_NCHAR) {
+    for(int32_t i = 0; i < pResultRowInfo->size; ++i) {
+      tfree(pResultRowInfo->pResult[i]->key);
     }
   }
   
-  tfree(pWindowResInfo->pResult);
+  tfree(pResultRowInfo->pResult);
 }
 
-void resetTimeWindowInfo(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowResInfo) {
-  if (pWindowResInfo == NULL || pWindowResInfo->capacity == 0) {
+void resetTimeWindowInfo(SQueryRuntimeEnv *pRuntimeEnv, SResultRowInfo *pResultRowInfo) {
+  if (pResultRowInfo == NULL || pResultRowInfo->capacity == 0) {
     return;
   }
 
-//  assert(pWindowResInfo->size == 1);
-
-  for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
-    SResultRow *pWindowRes = pWindowResInfo->pResult[i];
-    clearResultRow(pRuntimeEnv, pWindowRes, pWindowResInfo->type);
+  for (int32_t i = 0; i < pResultRowInfo->size; ++i) {
+    SResultRow *pWindowRes = pResultRowInfo->pResult[i];
+    clearResultRow(pRuntimeEnv, pWindowRes, pResultRowInfo->type);
 
     int32_t groupIndex = 0;
     int64_t uid = 0;
@@ -96,30 +93,30 @@ void resetTimeWindowInfo(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowR
     taosHashRemove(pRuntimeEnv->pResultRowHashTable, (const char *)pRuntimeEnv->keyBuf, GET_RES_WINDOW_KEY_LEN(sizeof(groupIndex)));
   }
   
-  pWindowResInfo->curIndex = -1;
-  pWindowResInfo->size = 0;
+  pResultRowInfo->curIndex = -1;
+  pResultRowInfo->size = 0;
   
-  pWindowResInfo->startTime = TSKEY_INITIAL_VAL;
-  pWindowResInfo->prevSKey = TSKEY_INITIAL_VAL;
+  pResultRowInfo->startTime = TSKEY_INITIAL_VAL;
+  pResultRowInfo->prevSKey = TSKEY_INITIAL_VAL;
 }
 
 void clearFirstNWindowRes(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
-  SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo;
-  if (pWindowResInfo == NULL || pWindowResInfo->capacity == 0 || pWindowResInfo->size == 0 || num == 0) {
+  SResultRowInfo *pResultRowInfo = &pRuntimeEnv->windowResInfo;
+  if (pResultRowInfo == NULL || pResultRowInfo->capacity == 0 || pResultRowInfo->size == 0 || num == 0) {
     return;
   }
   
-  int32_t numOfClosed = numOfClosedTimeWindow(pWindowResInfo);
+  int32_t numOfClosed = numOfClosedTimeWindow(pResultRowInfo);
   assert(num >= 0 && num <= numOfClosed);
 
-  int16_t type = pWindowResInfo->type;
+  int16_t type = pResultRowInfo->type;
   int64_t uid = getResultInfoUId(pRuntimeEnv);
 
   char    *key  = NULL;
   int16_t  bytes = -1;
 
   for (int32_t i = 0; i < num; ++i) {
-    SResultRow *pResult = pWindowResInfo->pResult[i];
+    SResultRow *pResult = pResultRowInfo->pResult[i];
     if (pResult->closed) {  // remove the window slot from hash table
       getResultRowKeyInfo(pResult, type, &key, &bytes);
       SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, key, bytes, uid);
@@ -129,23 +126,23 @@ void clearFirstNWindowRes(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
     }
   }
   
-  int32_t remain = pWindowResInfo->size - num;
+  int32_t remain = pResultRowInfo->size - num;
   
   // clear all the closed windows from the window list
   for (int32_t k = 0; k < remain; ++k) {
-    copyResultRow(pRuntimeEnv, pWindowResInfo->pResult[k], pWindowResInfo->pResult[num + k], type);
+    copyResultRow(pRuntimeEnv, pResultRowInfo->pResult[k], pResultRowInfo->pResult[num + k], type);
   }
   
   // move the unclosed window in the front of the window list
-  for (int32_t k = remain; k < pWindowResInfo->size; ++k) {
-    SResultRow *pWindowRes = pWindowResInfo->pResult[k];
-    clearResultRow(pRuntimeEnv, pWindowRes, pWindowResInfo->type);
+  for (int32_t k = remain; k < pResultRowInfo->size; ++k) {
+    SResultRow *pWindowRes = pResultRowInfo->pResult[k];
+    clearResultRow(pRuntimeEnv, pWindowRes, pResultRowInfo->type);
   }
   
-  pWindowResInfo->size = remain;
+  pResultRowInfo->size = remain;
 
-  for (int32_t k = 0; k < pWindowResInfo->size; ++k) {
-    SResultRow *pResult = pWindowResInfo->pResult[k];
+  for (int32_t k = 0; k < pResultRowInfo->size; ++k) {
+    SResultRow *pResult = pResultRowInfo->pResult[k];
     getResultRowKeyInfo(pResult, type, &key, &bytes);
     SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, key, bytes, uid);
 
@@ -153,43 +150,43 @@ void clearFirstNWindowRes(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
     assert(p != NULL); 
 
     int32_t  v = (*p - num);
-    assert(v >= 0 && v <= pWindowResInfo->size);
+    assert(v >= 0 && v <= pResultRowInfo->size);
 
     SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, key, bytes, uid);
     taosHashPut(pRuntimeEnv->pResultRowHashTable, pRuntimeEnv->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), (char *)&v, sizeof(int32_t));
   }
   
-  pWindowResInfo->curIndex = -1;
+  pResultRowInfo->curIndex = -1;
 }
 
 void clearClosedTimeWindow(SQueryRuntimeEnv *pRuntimeEnv) {
-  SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo;
-  if (pWindowResInfo == NULL || pWindowResInfo->capacity == 0 || pWindowResInfo->size == 0) {
+  SResultRowInfo *pResultRowInfo = &pRuntimeEnv->windowResInfo;
+  if (pResultRowInfo == NULL || pResultRowInfo->capacity == 0 || pResultRowInfo->size == 0) {
     return;
   }
   
-  int32_t numOfClosed = numOfClosedTimeWindow(pWindowResInfo);
+  int32_t numOfClosed = numOfClosedTimeWindow(pResultRowInfo);
   clearFirstNWindowRes(pRuntimeEnv, numOfClosed);
 }
 
-int32_t numOfClosedTimeWindow(SWindowResInfo *pWindowResInfo) {
+int32_t numOfClosedTimeWindow(SResultRowInfo *pResultRowInfo) {
   int32_t i = 0;
-  while (i < pWindowResInfo->size && pWindowResInfo->pResult[i]->closed) {
+  while (i < pResultRowInfo->size && pResultRowInfo->pResult[i]->closed) {
     ++i;
   }
   
   return i;
 }
 
-void closeAllTimeWindow(SWindowResInfo *pWindowResInfo) {
-  assert(pWindowResInfo->size >= 0 && pWindowResInfo->capacity >= pWindowResInfo->size);
+void closeAllTimeWindow(SResultRowInfo *pResultRowInfo) {
+  assert(pResultRowInfo->size >= 0 && pResultRowInfo->capacity >= pResultRowInfo->size);
   
-  for (int32_t i = 0; i < pWindowResInfo->size; ++i) {
-    if (pWindowResInfo->pResult[i]->closed) {
+  for (int32_t i = 0; i < pResultRowInfo->size; ++i) {
+    if (pResultRowInfo->pResult[i]->closed) {
       continue;
     }
     
-    pWindowResInfo->pResult[i]->closed = true;
+    pResultRowInfo->pResult[i]->closed = true;
   }
 }
 
@@ -198,41 +195,41 @@ void closeAllTimeWindow(SWindowResInfo *pWindowResInfo) {
  * 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) {
+void removeRedundantWindow(SResultRowInfo *pResultRowInfo, TSKEY lastKey, int32_t order) {
+  assert(pResultRowInfo->size >= 0 && pResultRowInfo->capacity >= pResultRowInfo->size);
+  if (pResultRowInfo->size <= 1) {
     return;
   }
 
   // get the result order
-  int32_t resultOrder = (pWindowResInfo->pResult[0]->win.skey < pWindowResInfo->pResult[1]->win.skey)? 1:-1;
+  int32_t resultOrder = (pResultRowInfo->pResult[0]->win.skey < pResultRowInfo->pResult[1]->win.skey)? 1:-1;
   if (order != resultOrder) {
     return;
   }
 
   int32_t i = 0;
   if (order == QUERY_ASC_FORWARD_STEP) {
-    TSKEY ekey = pWindowResInfo->pResult[i]->win.ekey;
-    while (i < pWindowResInfo->size && (ekey < lastKey)) {
+    TSKEY ekey = pResultRowInfo->pResult[i]->win.ekey;
+    while (i < pResultRowInfo->size && (ekey < lastKey)) {
       ++i;
     }
   } else if (order == QUERY_DESC_FORWARD_STEP) {
-    while (i < pWindowResInfo->size && (pWindowResInfo->pResult[i]->win.skey > lastKey)) {
+    while (i < pResultRowInfo->size && (pResultRowInfo->pResult[i]->win.skey > lastKey)) {
       ++i;
     }
   }
 
-  if (i < pWindowResInfo->size) {
-    pWindowResInfo->size = (i + 1);
+  if (i < pResultRowInfo->size) {
+    pResultRowInfo->size = (i + 1);
   }
 }
 
-bool isWindowResClosed(SWindowResInfo *pWindowResInfo, int32_t slot) {
-  return (getResultRow(pWindowResInfo, slot)->closed == true);
+bool isWindowResClosed(SResultRowInfo *pResultRowInfo, int32_t slot) {
+  return (getResultRow(pResultRowInfo, slot)->closed == true);
 }
 
-void closeTimeWindow(SWindowResInfo *pWindowResInfo, int32_t slot) {
-  getResultRow(pWindowResInfo, slot)->closed = true;
+void closeTimeWindow(SResultRowInfo *pResultRowInfo, int32_t slot) {
+  getResultRow(pResultRowInfo, slot)->closed = true;
 }
 
 void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pWindowRes, int16_t type) {

src/tsdb/src/tsdbFile.c

@@ -248,8 +248,13 @@ SFileGroup *tsdbCreateFGroupIfNeed(STsdbRepo *pRepo, char *dataDir, int fid) {
   if (pGroup == NULL) {  // if not exists, create one
     pFGroup->fileId = fid;
     for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) {
-      if (tsdbCreateFile(&pFGroup->files[type], pRepo, fid, type) < 0)
-        goto _err;
+      if (tsdbCreateFile(&pFGroup->files[type], pRepo, fid, type) < 0) {
+        for (int i = type; i >= 0; i--) {
+          remove(pFGroup->files[i].fname);
+        }
+
+        return NULL;
+      }
     }
 
     pthread_rwlock_wrlock(&pFileH->fhlock);
@@ -261,10 +266,6 @@ SFileGroup *tsdbCreateFGroupIfNeed(STsdbRepo *pRepo, char *dataDir, int fid) {
   }
 
   return pGroup;
-
-_err:
-  for (int type = 0; type < TSDB_FILE_TYPE_MAX; type++) tsdbDestroyFile(&pGroup->files[type]);
-  return NULL;
 }
 
 void tsdbInitFileGroupIter(STsdbFileH *pFileH, SFileGroupIter *pIter, int direction) {

src/util/src/hash.c

@@ -151,8 +151,9 @@ static void pushfrontNodeInEntryList(SHashEntry *pEntry, SHashNode *pNode);
  */
 
 SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTypeE type) {
-  if (capacity == 0 || fn == NULL) {
-    return NULL;
+  assert(fn != NULL);
+  if (capacity == 0) {
+    capacity = 4;
   }
 
   SHashObj *pHashObj = (SHashObj *)calloc(1, sizeof(SHashObj));

src/util/src/tconfig.c

@@ -19,9 +19,7 @@
 #include "taoserror.h"
 #include "tconfig.h"
 #include "tglobal.h"
-#include "tkey.h"
 #include "tulog.h"
-#include "tsocket.h"
 #include "tsystem.h"
 #include "tutil.h"
 

src/util/src/tref.c

@@ -329,7 +329,7 @@ void *taosIterateRef(int rsetId, int64_t rid) {
     pNode->count++;  // acquire it
     newP = pNode->p;  
     taosUnlockList(pSet->lockedBy+hash);
-    uTrace("rsetId:%d p:%p rid:%" PRId64 " is returned", rsetId, newP, rid);
+    uTrace("rsetId:%d p:%p rid:%" PRId64 " is returned", rsetId, newP, rid); 
   } else {
     uTrace("rsetId:%d the list is over", rsetId);
   }
@@ -423,24 +423,25 @@ static int taosDecRefCount(int rsetId, int64_t rid, int remove) {
       if (pNode->next) {
         pNode->next->prev = pNode->prev; 
       } 
-		
-      (*pSet->fp)(pNode->p); 
-
-      uTrace("rsetId:%d p:%p rid:%" PRId64 " is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count, pNode);
-      free(pNode);
       released = 1;
     } else {
-      uTrace("rsetId:%d p:%p rid:%" PRId64 " is released, count:%d", rsetId, pNode->p, rid, pNode->count);
+       uTrace("rsetId:%d p:%p rid:%" PRId64 " is released", rsetId, pNode->p, rid); 
     }
   } else {
-    uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to release/remove", rsetId, rid);
+    uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to release/remove", rsetId, rid); 
     terrno = TSDB_CODE_REF_NOT_EXIST;
     code = -1;
   }
 
   taosUnlockList(pSet->lockedBy+hash);
 
-  if (released) taosDecRsetCount(pSet);
+  if (released) {
+    uTrace("rsetId:%d p:%p rid:%" PRId64 " is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count, pNode);
+    (*pSet->fp)(pNode->p); 
+    free(pNode);
+
+    taosDecRsetCount(pSet);
+  } 
 
   return code;
 }

src/vnode/src/vnodeMain.c

@@ -267,16 +267,18 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
     return terrno;
   }
 
-  SCqCfg cqCfg = {0};
-  sprintf(cqCfg.user, "_root");
-  strcpy(cqCfg.pass, tsInternalPass);
-  strcpy(cqCfg.db, pVnode->db);
-  cqCfg.vgId = vnode;
-  cqCfg.cqWrite = vnodeWriteToCache;
-  pVnode->cq = cqOpen(pVnode, &cqCfg);
-  if (pVnode->cq == NULL) {
-    vnodeCleanUp(pVnode);
-    return terrno;
+  if (tsEnableStream) {
+    SCqCfg cqCfg = {0};
+    sprintf(cqCfg.user, "_root");
+    strcpy(cqCfg.pass, tsInternalPass);
+    strcpy(cqCfg.db, pVnode->db);
+    cqCfg.vgId = vnode;
+    cqCfg.cqWrite = vnodeWriteToCache;
+    pVnode->cq = cqOpen(pVnode, &cqCfg);
+    if (pVnode->cq == NULL) {
+      vnodeCleanUp(pVnode);
+      return terrno;
+    }
   }
 
   STsdbAppH appH = {0};

src/vnode/src/vnodeRead.c

@@ -275,41 +275,40 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
 
     vDebug("vgId:%d, QInfo:%p, dnode continues to exec query", pVnode->vgId, *qhandle);
 
+    // In the retrieve blocking model, only 50% CPU will be used in query processing
+    if (tsHalfCoresForQuery) {
+      qTableQuery(*qhandle);  // do execute query
+      qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, false);
+    } else {
+      bool freehandle = false;
+      bool buildRes = qTableQuery(*qhandle);  // do execute query
 
-#if _NON_BLOCKING_RETRIEVE
-    bool freehandle = false;
-    bool buildRes = qTableQuery(*qhandle);  // do execute query
-
-    // build query rsp, the retrieve request has reached here already
-    if (buildRes) {
-      // update the connection info according to the retrieve connection
-      pRead->rpcHandle = qGetResultRetrieveMsg(*qhandle);
-      assert(pRead->rpcHandle != NULL);
-
-      vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle,
-             pRead->rpcHandle);
+      // build query rsp, the retrieve request has reached here already
+      if (buildRes) {
+        // update the connection info according to the retrieve connection
+        pRead->rpcHandle = qGetResultRetrieveMsg(*qhandle);
+        assert(pRead->rpcHandle != NULL);
 
-      // set the real rsp error code
-      pRead->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pRead->rpcHandle);
+        vDebug("vgId:%d, QInfo:%p, start to build retrieval rsp after query paused, %p", pVnode->vgId, *qhandle,
+               pRead->rpcHandle);
 
-      // NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client
-      code = TSDB_CODE_QRY_HAS_RSP;
-    } else {
-      void* h1 = qGetResultRetrieveMsg(*qhandle);
-      assert(h1 == NULL);
+        // set the real rsp error code
+        pRead->code = vnodeDumpQueryResult(&pRead->rspRet, pVnode, qhandle, &freehandle, pRead->rpcHandle);
 
-      freehandle = qQueryCompleted(*qhandle);
-    }
+        // NOTE: set return code to be TSDB_CODE_QRY_HAS_RSP to notify dnode to return msg to client
+        code = TSDB_CODE_QRY_HAS_RSP;
+      } else {
+        void *h1 = qGetResultRetrieveMsg(*qhandle);
+        assert(h1 == NULL);
+        freehandle = qQueryCompleted(*qhandle);
+      }
 
-    // NOTE: if the qhandle is not put into vread queue or query is completed, free the qhandle.
-    // If the building of result is not required, simply free it. Otherwise, mandatorily free the qhandle
-    if (freehandle || (!buildRes)) {
-      qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, freehandle);
+      // NOTE: if the qhandle is not put into vread queue or query is completed, free the qhandle.
+      // If the building of result is not required, simply free it. Otherwise, mandatorily free the qhandle
+      if (freehandle || (!buildRes)) {
+        qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, freehandle);
+      }
     }
-#else
-    qTableQuery(*qhandle);  // do execute query
-    qReleaseQInfo(pVnode->qMgmt, (void **)&qhandle, false);
-#endif
   }
 
   return code;
@@ -375,14 +374,16 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
     freeHandle = true;
   } else {  // result is not ready, return immediately
     assert(buildRes == true);
-#if _NON_BLOCKING_RETRIEVE
-    if (!buildRes) {
-      assert(pRead->rpcHandle != NULL);
 
-      qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false);
-      return TSDB_CODE_QRY_NOT_READY;
+    // Only effects in the non-blocking model
+    if (!tsHalfCoresForQuery) {
+      if (!buildRes) {
+        assert(pRead->rpcHandle != NULL);
+
+        qReleaseQInfo(pVnode->qMgmt, (void **)&handle, false);
+        return TSDB_CODE_QRY_NOT_READY;
+      }
     }
-#endif
 
     // ahandle is the sqlObj pointer
     code = vnodeDumpQueryResult(pRet, pVnode, handle, &freeHandle, pRead->rpcHandle);

tests/examples/JDBC/JDBCDemo/pom.xml

@@ -56,7 +56,7 @@
         <dependency>
             <groupId>com.taosdata.jdbc</groupId>
             <artifactId>taos-jdbcdriver</artifactId>
-            <version>2.0.8</version>
+            <version>2.0.12</version>
         </dependency>
         <dependency>
             <groupId>log4j</groupId>

tests/examples/JDBC/JDBCDemo/readme.md

@@ -6,10 +6,24 @@ TDengine's JDBC demo project is organized in a Maven way so that users can easil
 Make sure you have already installed a tdengine client on your current develop environment.
 Download the tdengine package on our website: ``https://www.taosdata.com/cn/all-downloads/`` and install the client.
 
+## How to run jdbcChecker
+<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.JdbcChecker" -Dexec.args="-host localhost"</pre>
+
+## How to run jdbcTaosDemo
+run command:
+<pre> mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo"</pre>
+and run with your customed args
+<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo" -Dexec.args="-host localhost"</pre>
+
 ## Compile the Demo Code and Run It
+
 To compile the demo project, go to the source directory ``TDengine/tests/examples/JDBC/JDBCDemo`` and execute
-<pre>mvn clean package assembly:single</pre>
+
+<pre>
+mvn clean package assembly:single
+</pre>
+
 The ``pom.xml`` is configured to package all the dependencies into one executable jar file.
 
 To run it, go to ``examples/JDBC/JDBCDemo/target`` and execute
-<pre>java -jar jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host localhost</pre>
+<pre>java -jar jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host localhost</pre>
\ No newline at end of file

tests/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/jdbcTaosdemo/domain/JdbcTaosdemoConfig.java

@@ -14,9 +14,9 @@ public final class JdbcTaosdemoConfig {
     //Destination database. Default is 'test'
     private String dbName = "test";
     //keep
-    private int keep = 3650;
+    private int keep = 36500;
     //days
-    private int days = 10;
+    private int days = 120;
 
     //Super table Name. Default is 'meters'
     private String stbName = "meters";

tests/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/jdbcTaosdemo/task/InsertTableTask.java

@@ -41,7 +41,7 @@ public class InsertTableTask implements Runnable {
                 long ts = start.toEpochMilli() + (j * timeGap);
                 // insert data into echo table
                 for (int i = startTbIndex; i < startTbIndex + tableNumber; i++) {
-                    String sql = SqlSpeller.insertOneRowSQL(config.getDbName(), config.getTbPrefix(), i + 1, ts);
+                    String sql = SqlSpeller.insertBatchSizeRowsSQL(config.getDbName(), config.getTbPrefix(), i + 1, ts, config.getNumberOfRecordsPerRequest());
                     logger.info(Thread.currentThread().getName() + ">>> " + sql);
                     Statement statement = connection.createStatement();
                     statement.execute(sql);

tests/pytest/client/alterDatabase.py

@@ -32,9 +32,9 @@ class TDTestCase:
         tdSql.query("show databases")
         tdSql.checkData(0, 14, 2)
 
-        tdSql.execute("alter database db keep 365")
+        tdSql.execute("alter database db keep 365,365,365")
         tdSql.query("show databases")
-        tdSql.checkData(0, 7, "3650,3650,365")
+        tdSql.checkData(0, 7, "365,365,365")
 
         tdSql.execute("alter database db quorum 2")
         tdSql.query("show databases")

tests/pytest/crash_gen/db.py

@@ -78,7 +78,7 @@ class DbConn:
         if nRows != 1:
             raise taos.error.ProgrammingError(
                 "Unexpected result for query: {}, rows = {}".format(sql, nRows), 
-                (0x991 if nRows==0 else 0x992)
+                (CrashGenError.INVALID_EMPTY_RESULT if nRows==0 else CrashGenError.INVALID_MULTIPLE_RESULT)
             )
         if self.getResultRows() != 1 or self.getResultCols() != 1:
             raise RuntimeError("Unexpected result set for query: {}".format(sql))
@@ -349,7 +349,8 @@ class DbConnNative(DbConn):
 
     def execute(self, sql):
         if (not self.isOpen):
-            raise RuntimeError("Cannot execute database commands until connection is open")
+            raise CrashGenError(
+                "Cannot exec SQL unless db connection is open", CrashGenError.DB_CONNECTION_NOT_OPEN)
         Logging.debug("[SQL] Executing SQL: {}".format(sql))
         self._lastSql = sql
         nRows = self._tdSql.execute(sql)
@@ -360,8 +361,8 @@ class DbConnNative(DbConn):
 
     def query(self, sql):  # return rows affected
         if (not self.isOpen):
-            raise RuntimeError(
-                "Cannot query database until connection is open")
+            raise CrashGenError(
+                "Cannot query database until connection is open, restarting?", CrashGenError.DB_CONNECTION_NOT_OPEN)
         Logging.debug("[SQL] Executing SQL: {}".format(sql))
         self._lastSql = sql
         nRows = self._tdSql.query(sql)

tests/pytest/crash_gen/misc.py

@@ -3,14 +3,20 @@ import random
 import logging
 import os
 
+import taos
 
-class CrashGenError(Exception):
-    def __init__(self, msg=None, errno=None):
-        self.msg = msg
-        self.errno = errno
 
-    def __str__(self):
-        return self.msg
+class CrashGenError(taos.error.ProgrammingError):
+    INVALID_EMPTY_RESULT    = 0x991
+    INVALID_MULTIPLE_RESULT = 0x992
+    DB_CONNECTION_NOT_OPEN  = 0x993
+    # def __init__(self, msg=None, errno=None):
+    #     self.msg = msg
+    #     self.errno = errno
+
+    # def __str__(self):
+    #     return self.msg
+    pass
 
 
 class LoggingFilter(logging.Filter):
@@ -168,6 +174,7 @@ class Progress:
     SERVICE_RECONNECT_FAILURE   = 6
     SERVICE_START_NAP           = 7
     CREATE_TABLE_ATTEMPT        = 8
+    QUERY_GROUP_BY              = 9
 
     tokens = {
         STEP_BOUNDARY:      '.',
@@ -178,7 +185,8 @@ class Progress:
         SERVICE_RECONNECT_SUCCESS:  '.r>',
         SERVICE_RECONNECT_FAILURE:  '.xr>',
         SERVICE_START_NAP:           '_zz',
-        CREATE_TABLE_ATTEMPT: '_c',
+        CREATE_TABLE_ATTEMPT:       'c',
+        QUERY_GROUP_BY:             'g',
     }
 
     @classmethod

tests/pytest/crash_gen/service_manager.py

@@ -51,10 +51,12 @@ class TdeInstance():
     def prepareGcovEnv(cls, env):
         # Ref: https://gcc.gnu.org/onlinedocs/gcc/Cross-profiling.html
         bPath = cls._getBuildPath() # build PATH
-        numSegments = len(bPath.split('/')) - 1 # "/x/TDengine/build" should yield 3
-        numSegments = numSegments - 1 # DEBUG only
-        env['GCOV_PREFIX'] = bPath + '/svc_gcov'
+        numSegments = len(bPath.split('/')) # "/x/TDengine/build" should yield 3
+        # numSegments += 2 # cover "/src" after build
+        # numSegments = numSegments - 1 # DEBUG only
+        env['GCOV_PREFIX'] = bPath + '/src_s' # Server side source
         env['GCOV_PREFIX_STRIP'] = str(numSegments) # Strip every element, plus, ENV needs strings
+        # VERY VERY important note: GCOV data collection NOT effective upon SIG_KILL
         Logging.info("Preparing GCOV environement to strip {} elements and use path: {}".format(
             numSegments, env['GCOV_PREFIX'] ))
 
@@ -258,14 +260,15 @@ class TdeSubProcess:
         TdeInstance.prepareGcovEnv(myEnv)
 
         # print(myEnv)
-        # print(myEnv.items())
+        # print("Starting TDengine with env: ", myEnv.items())
         # print("Starting TDengine via Shell: {}".format(cmdLineStr))
 
         useShell = True    
         self.subProcess = subprocess.Popen(
-            ' '.join(cmdLine) if useShell else cmdLine,
-            shell=useShell,
-            # svcCmdSingle, shell=True, # capture core dump?
+            # ' '.join(cmdLine) if useShell else cmdLine,
+            # shell=useShell,
+            ' '.join(cmdLine),
+            shell=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             # bufsize=1, # not supported in binary mode
@@ -273,7 +276,8 @@ class TdeSubProcess:
             env=myEnv
             )  # had text=True, which interferred with reading EOF
 
-    STOP_SIGNAL = signal.SIGKILL # What signal to use (in kill) to stop a taosd process?
+    STOP_SIGNAL = signal.SIGKILL # signal.SIGKILL/SIGINT # What signal to use (in kill) to stop a taosd process?
+    SIG_KILL_RETCODE = 137 # ref: https://stackoverflow.com/questions/43268156/process-finished-with-exit-code-137-in-pycharm
 
     def stop(self):
         """
@@ -320,8 +324,12 @@ class TdeSubProcess:
         retCode = self.subProcess.returncode # should always be there
         # May throw subprocess.TimeoutExpired exception above, therefore
         # The process is guranteed to have ended by now
-        self.subProcess = None        
-        if retCode != 0: # != (- signal.SIGINT):
+        self.subProcess = None       
+        if retCode == self.SIG_KILL_RETCODE:
+            Logging.info("TSP.stop(): sub proc KILLED, as expected")
+        elif retCode == (- self.STOP_SIGNAL):
+            Logging.info("TSP.stop(), sub process STOPPED, as expected")
+        elif retCode != 0: # != (- signal.SIGINT):
             Logging.error("TSP.stop(): Failed to stop sub proc properly w/ SIG {}, retCode={}".format(
                 self.STOP_SIGNAL, retCode))
         else:

tests/pytest/fulltest.sh

@@ -19,6 +19,7 @@ python3 ./test.py -f insert/randomNullCommit.py
 python3 insert/retentionpolicy.py
 python3 ./test.py -f insert/alterTableAndInsert.py
 python3 ./test.py -f insert/insertIntoTwoTables.py
+python3 ./test.py -f insert/before_1970.py
 
 python3 ./test.py -f table/alter_wal0.py
 python3 ./test.py -f table/column_name.py
@@ -228,6 +229,7 @@ python3 ./test.py -f update/merge_commit_data2.py
 python3 ./test.py -f update/merge_commit_data2_update0.py
 python3 ./test.py -f update/merge_commit_last-0.py
 python3 ./test.py -f update/merge_commit_last.py
+python3 ./test.py -f update/bug_td2279.py
 
 # wal
 python3 ./test.py -f wal/addOldWalTest.py
\ No newline at end of file

tests/script/general/parser/first_last_query.sim

@@ -266,4 +266,6 @@ endi
 if $data14 != @test2@ then
   print expect test2 , actual: $data14
   return -1
-endi
+endi
\ No newline at end of file
+
+sql drop table stest
\ No newline at end of file