Merge branch 'develop' into hotfix/TD-919

src/client/src/tscAsync.c

@@ -564,8 +564,8 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
     tscDebug("%p stream:%p meta is updated, start new query, command:%d", pSql, pSql->pStream, pSql->cmd.command);
     if (!pSql->cmd.parseFinished) {
       tsParseSql(pSql, false);
-      sem_post(&pSql->rspSem);
     }
+    (*pSql->fp)(pSql->param, pSql, code);
 
     return;
   }

src/client/src/tscSQLParser.c

@@ -4642,21 +4642,24 @@ typedef struct SDNodeDynConfOption {
 } SDNodeDynConfOption;
 
 
-int32_t validateEp(char* ep) {
+int32_t validateEp(char* ep) {  
   char buf[TSDB_EP_LEN + 1] = {0};
   tstrncpy(buf, ep, TSDB_EP_LEN);
 
-  char *pos = strchr(buf, ':');
-  if (NULL == pos) {   
-    return TSDB_CODE_TSC_INVALID_SQL;
+  char* pos = strchr(buf, ':');
+  if (NULL == pos) {
+    int32_t val = strtol(ep, NULL, 10);
+    if (val <= 0 || val > 65536) {
+      return TSDB_CODE_TSC_INVALID_SQL;
+    }
+  } else {
+    uint16_t port = atoi(pos + 1);
+    if (0 == port) {
+      return TSDB_CODE_TSC_INVALID_SQL;
+    }
   }
-  
-  uint16_t port = atoi(pos+1);
-  if (0 == port) {   
-    return TSDB_CODE_TSC_INVALID_SQL;
-  }  
 
-  return TSDB_CODE_SUCCESS;   
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t validateDNodeConfig(tDCLSQL* pOptions) {
@@ -4664,13 +4667,13 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
     return TSDB_CODE_TSC_INVALID_SQL;
   }
 
-  const int DNODE_DYNAMIC_CFG_OPTIONS_SIZE = 17;
+  const int DNODE_DYNAMIC_CFG_OPTIONS_SIZE = 19;
   const SDNodeDynConfOption DNODE_DYNAMIC_CFG_OPTIONS[] = {
       {"resetLog", 8},       {"resetQueryCache", 15},  {"debugFlag", 9},     {"mDebugFlag", 10},
       {"dDebugFlag", 10},    {"sdbDebugFlag", 12},     {"vDebugFlag", 10},   {"cDebugFlag", 10},
       {"httpDebugFlag", 13}, {"monitorDebugFlag", 16}, {"rpcDebugFlag", 12}, {"uDebugFlag", 10},
       {"tmrDebugFlag", 12},  {"qDebugflag", 10},       {"sDebugflag", 10},   {"tsdbDebugFlag", 13},
-      {"monitor", 7}};
+      {"mqttDebugFlag", 13}, {"wDebugFlag", 10},       {"monitor", 7}};
 
   SSQLToken* pOptionToken = &pOptions->a[1];
 
@@ -4694,7 +4697,7 @@ int32_t validateDNodeConfig(tDCLSQL* pOptions) {
     SSQLToken* pValToken = &pOptions->a[2];
 
     int32_t val = strtol(pValToken->z, NULL, 10);
-    if (val < 131 || val > 199) {
+    if (val < 0 || val > 256) {
       /* options value is out of valid range */
       return TSDB_CODE_TSC_INVALID_SQL;
     }

src/client/src/tscStream.c

@@ -70,6 +70,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
   SSqlObj *   pSql = pStream->pSql;
 
   pSql->fp = tscProcessStreamQueryCallback;
+  pSql->fetchFp = tscProcessStreamQueryCallback;
   pSql->param = pStream;
   pSql->res.completed = false;
   
@@ -471,6 +472,41 @@ static void setErrorInfo(SSqlObj* pSql, int32_t code, char* info) {
   }
 }
 
+static void tscCreateStream(void *param, TAOS_RES *res, int code) {
+  SSqlStream* pStream = (SSqlStream*)param;
+  SSqlObj* pSql = pStream->pSql;
+  SSqlCmd* pCmd = &pSql->cmd;
+
+  if (code != TSDB_CODE_SUCCESS) {
+    setErrorInfo(pSql, code, pCmd->payload);
+    tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, pSql->sqlstr, pCmd->payload, code);
+    pStream->fp(pStream->param, NULL, NULL);
+    return;
+  }
+
+  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
+  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+  STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
+  
+  pStream->isProject = isProjectStream(pQueryInfo);
+  pStream->precision = tinfo.precision;
+
+  pStream->ctime = taosGetTimestamp(pStream->precision);
+  pStream->etime = pQueryInfo->window.ekey;
+
+  tscAddIntoStreamList(pStream);
+
+  tscSetSlidingWindowInfo(pSql, pStream);
+  pStream->stime = tscGetStreamStartTimestamp(pSql, pStream, pStream->stime);
+
+  int64_t starttime = tscGetLaunchTimestamp(pStream);
+  pCmd->command = TSDB_SQL_SELECT;
+  taosTmrReset(tscProcessStreamTimer, starttime, pStream, tscTmr, &pStream->pTimer);
+
+  tscDebug("%p stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql,
+           pStream, pTableMetaInfo->name, pStream->interval, pStream->slidingTime, starttime, pSql->sqlstr);
+}
+
 TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
                               int64_t stime, void *param, void (*callback)(void *)) {
   STscObj *pObj = (STscObj *)taos;
@@ -482,7 +518,6 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   }
 
   pSql->signature = pSql;
-  pSql->param = pSql;
   pSql->pTscObj = pObj;
 
   SSqlCmd *pCmd = &pSql->cmd;
@@ -494,7 +529,14 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
     tscFreeSqlObj(pSql);
     return NULL;
   }
+
+  pStream->stime = stime;
+  pStream->fp = fp;
+  pStream->callback = callback;
+  pStream->param = param;
+  pStream->pSql = pSql;
   pSql->pStream = pStream;
+  pSql->param = pStream;
 
   pSql->sqlstr = calloc(1, strlen(sqlstr) + 1);
   if (pSql->sqlstr == NULL) {
@@ -507,45 +549,18 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   tscDebugL("%p SQL: %s", pSql, pSql->sqlstr);
   tsem_init(&pSql->rspSem, 0, 0);
 
+  pSql->fp = tscCreateStream;
+  pSql->fetchFp = tscCreateStream;
   int32_t code = tsParseSql(pSql, true);
-  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
-    sem_wait(&pSql->rspSem);
-  }
-
-  if (pRes->code != TSDB_CODE_SUCCESS) {
-    setErrorInfo(pSql, pRes->code, pCmd->payload);
-
-    tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
+  if (code == TSDB_CODE_SUCCESS) {
+    tscCreateStream(pStream, pSql, code);
+  } else if (code != TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    tscError("%p open stream failed, sql:%s, code:%s", pSql, sqlstr, tstrerror(pRes->code));
     tscFreeSqlObj(pSql);
+    free(pStream);
     return NULL;
   }
 
-  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
-  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
-  STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
-  
-  pStream->isProject = isProjectStream(pQueryInfo);
-  pStream->fp = fp;
-  pStream->callback = callback;
-  pStream->param = param;
-  pStream->pSql = pSql;
-  pStream->precision = tinfo.precision;
-
-  pStream->ctime = taosGetTimestamp(pStream->precision);
-  pStream->etime = pQueryInfo->window.ekey;
-
-  tscAddIntoStreamList(pStream);
-
-  tscSetSlidingWindowInfo(pSql, pStream);
-  pStream->stime = tscGetStreamStartTimestamp(pSql, pStream, stime);
-
-  int64_t starttime = tscGetLaunchTimestamp(pStream);
-  pCmd->command = TSDB_SQL_SELECT;
-  taosTmrReset(tscProcessStreamTimer, starttime, pStream, tscTmr, &pStream->pTimer);
-
-  tscDebug("%p stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql,
-           pStream, pTableMetaInfo->name, pStream->interval, pStream->slidingTime, starttime, sqlstr);
-
   return pStream;
 }
 

src/cq/src/cqMain.c

@@ -23,6 +23,7 @@
 #include "taos.h"
 #include "taosdef.h"
 #include "taosmsg.h"
+#include "ttimer.h"
 #include "tcq.h"
 #include "tdataformat.h"
 #include "tglobal.h"
@@ -45,10 +46,12 @@ typedef struct {
   struct SCqObj *pHead;
   void    *dbConn;
   int      master;
+  void    *tmrCtrl;
   pthread_mutex_t mutex;
 } SCqContext;
 
 typedef struct SCqObj {
+  tmr_h          tmrId;
   uint64_t       uid;
   int32_t        tid;      // table ID
   int            rowSize;  // bytes of a row
@@ -66,13 +69,14 @@ 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) {
-  
   SCqContext *pContext = calloc(sizeof(SCqContext), 1);
   if (pContext == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return NULL;
   }
 
+  pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ");
+
   tstrncpy(pContext->user, pCfg->user, sizeof(pContext->user));
   tstrncpy(pContext->pass, pCfg->pass, sizeof(pContext->pass));
   const char* db = pCfg->db;
@@ -99,6 +103,9 @@ void cqClose(void *handle) {
   SCqContext *pContext = handle;
   if (handle == NULL) return;
 
+  taosTmrCleanUp(pContext->tmrCtrl);
+  pContext->tmrCtrl = NULL;
+
   // stop all CQs
   cqStop(pContext);
 
@@ -154,8 +161,10 @@ void cqStop(void *handle) {
       taos_close_stream(pObj->pStream);
       pObj->pStream = NULL;
       cTrace("vgId:%d, id:%d CQ:%s is closed", pContext->vgId, pObj->tid, pObj->sqlStr);
+    } else {
+      taosTmrStop(pObj->tmrId);
+      pObj->tmrId = 0;
     }
-
     pObj = pObj->next;
   }
 
@@ -211,8 +220,13 @@ void cqDrop(void *handle) {
   }
 
   // free the resources associated
-  if (pObj->pStream) taos_close_stream(pObj->pStream);
-  pObj->pStream = NULL;
+  if (pObj->pStream) {
+    taos_close_stream(pObj->pStream);
+    pObj->pStream = NULL;
+  } else {
+    taosTmrStop(pObj->tmrId);
+    pObj->tmrId = 0;
+  }
 
   cTrace("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr); 
   tdFreeSchema(pObj->pSchema);
@@ -222,18 +236,30 @@ void cqDrop(void *handle) {
   pthread_mutex_unlock(&pContext->mutex);
 }
 
-static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
+static void cqProcessCreateTimer(void *param, void *tmrId) {
+  SCqObj* pObj = (SCqObj*)param;
+  SCqContext* pContext = pObj->pContext;
+
   if (pContext->dbConn == NULL) {
     pContext->dbConn = taos_connect("localhost", pContext->user, pContext->pass, pContext->db, 0);
     if (pContext->dbConn == NULL) {
       cError("vgId:%d, failed to connect to TDengine(%s)", pContext->vgId, tstrerror(terrno));
-      return;
     }
   }
+  
+  cqCreateStream(pContext, pObj);
+}
 
-  int64_t lastKey = 0;
+static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
   pObj->pContext = pContext;
-  pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL);
+
+  if (pContext->dbConn == NULL) {
+    pObj->tmrId = taosTmrStart(cqProcessCreateTimer, 1000, pObj, pContext->tmrCtrl);
+    return;
+  }
+  pObj->tmrId = 0;
+
+  pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL);
   if (pObj->pStream) {
     pContext->num++;
     cTrace("vgId:%d, id:%d CQ:%s is openned", pContext->vgId, pObj->tid, pObj->sqlStr);

src/dnode/inc/dnodeModule.h

@@ -22,7 +22,6 @@ extern "C" {
 
 int32_t dnodeInitModules();
 void    dnodeStartModules();
-void    dnodeStartStream();
 void    dnodeCleanupModules();
 void    dnodeProcessModuleStatus(uint32_t moduleStatus);
 

src/dnode/src/dnodeMain.c

@@ -123,7 +123,6 @@ int32_t dnodeInitSystem() {
 
   dnodeStartModules();
   dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING);
-  dnodeStartStream();
 
   dInfo("TDengine is initialized successfully");
 

src/dnode/src/dnodeMgmt.c

@@ -354,23 +354,6 @@ static int32_t dnodeOpenVnodes() {
   return TSDB_CODE_SUCCESS;
 }
 
-void dnodeStartStream() {
-  int32_t vnodeList[TSDB_MAX_VNODES] = {0};
-  int32_t numOfVnodes = 0;
-  int32_t status = vnodeGetVnodeList(vnodeList, &numOfVnodes);
-
-  if (status != TSDB_CODE_SUCCESS) {
-    dInfo("get dnode list failed");
-    return;
-  }
-
-  for (int32_t i = 0; i < numOfVnodes; ++i) {
-    vnodeStartStream(vnodeList[i]);
-  }
-
-  dInfo("streams started");
-}
-
 static void dnodeCloseVnodes() {
   int32_t vnodeList[TSDB_MAX_VNODES]= {0};
   int32_t numOfVnodes = 0;
@@ -723,6 +706,7 @@ static void dnodeSendStatusMsg(void *handle, void *tmrId) {
 
   // fill cluster cfg parameters
   pStatus->clusterCfg.numOfMnodes        = htonl(tsNumOfMnodes);
+  pStatus->clusterCfg.enableBalance      = htonl(tsEnableBalance);
   pStatus->clusterCfg.mnodeEqualVnodeNum = htonl(tsMnodeEqualVnodeNum);
   pStatus->clusterCfg.offlineThreshold   = htonl(tsOfflineThreshold);
   pStatus->clusterCfg.statusInterval     = htonl(tsStatusInterval);

src/inc/taosmsg.h

@@ -54,12 +54,11 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CREATE_TABLE, "create-table" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_TABLE, "drop-table" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_TABLE, "alter-table" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CREATE_VNODE, "create-vnode" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_VNODE, "alter-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_VNODE, "drop-vnode" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_STABLE, "drop-stable" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_STREAM, "alter-stream" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CONFIG_DNODE, "config-dnode" )
-TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY4, "dummy4" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_VNODE, "alter-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY5, "dummy5" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY6, "dummy6" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY7, "dummy7" )
@@ -564,15 +563,16 @@ typedef struct {
 
 typedef struct {
   int32_t  numOfMnodes;               // tsNumOfMnodes
+  int32_t  enableBalance;             // tsEnableBalance
   int32_t  mnodeEqualVnodeNum;        // tsMnodeEqualVnodeNum
   int32_t  offlineThreshold;          // tsOfflineThreshold
   int32_t  statusInterval;            // tsStatusInterval
+  int32_t  maxtablesPerVnode;
+  int32_t  maxVgroupsPerDb;
   char     arbitrator[TSDB_EP_LEN];   // tsArbitrator
   char     timezone[64];              // tsTimezone
   char     locale[TSDB_LOCALE_LEN];   // tsLocale
   char     charset[TSDB_LOCALE_LEN];  // tsCharset
-  int32_t  maxtablesPerVnode;
-  int32_t  maxVgroupsPerDb;
 } SClusterCfg;
 
 typedef struct {

src/inc/tsdb.h

@@ -117,7 +117,6 @@ int   tsdbCreateTable(TSDB_REPO_T *repo, STableCfg *pCfg);
 int   tsdbDropTable(TSDB_REPO_T *pRepo, STableId tableId);
 int   tsdbUpdateTableTagValue(TSDB_REPO_T *repo, SUpdateTableTagValMsg *pMsg);
 TSKEY tsdbGetTableLastKey(TSDB_REPO_T *repo, uint64_t uid);
-void  tsdbStartStream(TSDB_REPO_T *repo);
 
 uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_t eindex, int32_t *size);
 

src/inc/vnode.h

@@ -44,7 +44,6 @@ typedef struct {
 int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg);
 int32_t vnodeDrop(int32_t vgId);
 int32_t vnodeOpen(int32_t vgId, char *rootDir);
-int32_t vnodeStartStream(int32_t vgId);
 int32_t vnodeAlter(void *pVnode, SMDCreateVnodeMsg *pVnodeCfg);
 int32_t vnodeClose(int32_t vgId);
 

src/mnode/src/mnodeDnode.c

@@ -269,18 +269,37 @@ void mnodeUpdateDnode(SDnodeObj *pDnode) {
 }
 
 static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
+  if (strcmp(pMsg->pUser->user, TSDB_DEFAULT_USER) != 0) {
+    mError("failed to cfg dnode, no rights");
+    return TSDB_CODE_MND_NO_RIGHTS;
+  }
+  
   SCMCfgDnodeMsg *pCmCfgDnode = pMsg->rpcMsg.pCont;
   if (pCmCfgDnode->ep[0] == 0) {
-    strcpy(pCmCfgDnode->ep, tsLocalEp);
-  } else {
-    // TODO temporary disabled for compiling: strcpy(pCmCfgDnode->ep, pCmCfgDnode->ep); 
+    tstrncpy(pCmCfgDnode->ep, tsLocalEp, TSDB_EP_LEN);
+  } 
+
+  int32_t dnodeId = 0;
+  char* pos = strchr(pCmCfgDnode->ep, ':');
+  if (NULL == pos) {
+    dnodeId = strtol(pCmCfgDnode->ep, NULL, 10);
+    if (dnodeId <= 0 || dnodeId > 65536) {
+      mError("failed to cfg dnode, invalid dnodeId:%s", pCmCfgDnode->ep);
+      return TSDB_CODE_MND_DNODE_NOT_EXIST;
+    }
   }
 
-  if (strcmp(pMsg->pUser->user, TSDB_DEFAULT_USER) != 0) {
-    return TSDB_CODE_MND_NO_RIGHTS;
+  SRpcIpSet ipSet = mnodeGetIpSetFromIp(pCmCfgDnode->ep);
+  if (dnodeId != 0) {
+    SDnodeObj *pDnode = mnodeGetDnode(dnodeId);
+    if (pDnode == NULL) {
+      mError("failed to cfg dnode, invalid dnodeId:%d", dnodeId);
+      return TSDB_CODE_MND_DNODE_NOT_EXIST;
+    }
+    ipSet = mnodeGetIpSetFromIp(pDnode->dnodeEp);
+    mnodeDecDnodeRef(pDnode);
   }
 
-  SRpcIpSet ipSet = mnodeGetIpSetFromIp(pCmCfgDnode->ep);
   SMDCfgDnodeMsg *pMdCfgDnode = rpcMallocCont(sizeof(SMDCfgDnodeMsg));
   strcpy(pMdCfgDnode->ep, pCmCfgDnode->ep);
   strcpy(pMdCfgDnode->config, pCmCfgDnode->config);
@@ -292,9 +311,9 @@ static int32_t mnodeProcessCfgDnodeMsg(SMnodeMsg *pMsg) {
     .pCont = pMdCfgDnode,
     .contLen = sizeof(SMDCfgDnodeMsg)
   };
-  dnodeSendMsgToDnode(&ipSet, &rpcMdCfgDnodeMsg);
 
   mInfo("dnode:%s, is configured by %s", pCmCfgDnode->ep, pMsg->pUser->user);
+  dnodeSendMsgToDnode(&ipSet, &rpcMdCfgDnodeMsg);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -305,6 +324,7 @@ static void mnodeProcessCfgDnodeMsgRsp(SRpcMsg *rpcMsg) {
 
 static bool mnodeCheckClusterCfgPara(const SClusterCfg *clusterCfg) {
   if (clusterCfg->numOfMnodes        != htonl(tsNumOfMnodes))        return false;
+  if (clusterCfg->enableBalance      != htonl(tsEnableBalance))        return false;
   if (clusterCfg->mnodeEqualVnodeNum != htonl(tsMnodeEqualVnodeNum)) return false;
   if (clusterCfg->offlineThreshold   != htonl(tsOfflineThreshold))   return false;
   if (clusterCfg->statusInterval     != htonl(tsStatusInterval))     return false;

src/mnode/src/mnodeVgroup.c

@@ -593,7 +593,7 @@ static int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *p
 
   pShow->bytes[cols] = 4;
   pSchema[cols].type = TSDB_DATA_TYPE_INT;
-  strcpy(pSchema[cols].name, "maxTables");
+  strcpy(pSchema[cols].name, "onlineVnodes");
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
@@ -692,8 +692,15 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
     *(int32_t *)pWrite = taosIdPoolMaxSize(pVgroup->idPool);
     cols++;
 
+    int32_t onlineVnodes = 0;
+    for (int32_t i = 0; i < pShow->maxReplica; ++i) {
+      if (pVgroup->vnodeGid[i].role == TAOS_SYNC_ROLE_SLAVE || pVgroup->vnodeGid[i].role == TAOS_SYNC_ROLE_MASTER) {
+        onlineVnodes++;
+      }
+    }
+
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    *(int32_t *)pWrite = tsMaxTablePerVnode;
+    *(int32_t *)pWrite = onlineVnodes;
     cols++;
 
     for (int32_t i = 0; i < pShow->maxReplica; ++i) {

src/tsdb/src/tsdbMain.c

@@ -69,6 +69,8 @@ static int         tsdbEncodeCfg(void **buf, STsdbCfg *pCfg);
 static void *      tsdbDecodeCfg(void *buf, STsdbCfg *pCfg);
 static int         tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable);
 static int         tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg);
+static void        tsdbStartStream(STsdbRepo *pRepo);
+static void        tsdbStopStream(STsdbRepo *pRepo);
 
 // Function declaration
 int32_t tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg) {
@@ -127,6 +129,7 @@ TSDB_REPO_T *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH) {
     goto _err;
   }
 
+  tsdbStartStream(pRepo);
   // pRepo->state = TSDB_REPO_STATE_ACTIVE;
 
   tsdbDebug("vgId:%d open tsdb repository succeed!", REPO_ID(pRepo));
@@ -145,6 +148,8 @@ void tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
   int        vgId = REPO_ID(pRepo);
 
+  tsdbStopStream(repo);
+
   if (toCommit) {
     tsdbAsyncCommit(pRepo);
     if (pRepo->commit) pthread_join(pRepo->commitThread, NULL);
@@ -265,19 +270,6 @@ uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_
   return magic;
 }
 
-void tsdbStartStream(TSDB_REPO_T *repo) {
-  STsdbRepo *pRepo = (STsdbRepo *)repo;
-  STsdbMeta *pMeta = pRepo->tsdbMeta;
-
-  for (int i = 0; i < pRepo->config.maxTables; i++) {
-    STable *pTable = pMeta->tables[i];
-    if (pTable && pTable->type == TSDB_STREAM_TABLE) {
-      pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, TABLE_UID(pTable), TABLE_TID(pTable), pTable->sql,
-                                                     tsdbGetTableSchemaImpl(pTable, false, false, -1));
-    }
-  }
-}
-
 STsdbCfg *tsdbGetCfg(const TSDB_REPO_T *repo) {
   ASSERT(repo != NULL);
   return &((STsdbRepo *)repo)->config;
@@ -1120,4 +1112,27 @@ TSKEY tsdbGetTableLastKey(TSDB_REPO_T *repo, uint64_t uid) {
   return TSDB_GET_TABLE_LAST_KEY(pTable);
 }
 
-#endif
\ No newline at end of file
+#endif
+
+static void tsdbStartStream(STsdbRepo *pRepo) {
+  STsdbMeta *pMeta = pRepo->tsdbMeta;
+
+  for (int i = 0; i < pRepo->config.maxTables; i++) {
+    STable *pTable = pMeta->tables[i];
+    if (pTable && pTable->type == TSDB_STREAM_TABLE) {
+      pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, TABLE_UID(pTable), TABLE_TID(pTable), pTable->sql,
+                                                     tsdbGetTableSchemaImpl(pTable, false, false, -1));
+    }
+  }
+}
+
+static void tsdbStopStream(STsdbRepo *pRepo) {
+  STsdbMeta *pMeta = pRepo->tsdbMeta;
+
+  for (int i = 0; i < pRepo->config.maxTables; i++) {
+    STable *pTable = pMeta->tables[i];
+    if (pTable && pTable->type == TSDB_STREAM_TABLE) {
+      (*pRepo->appH.cqDropFunc)(pTable->cqhandle);
+    }
+  }
+}

src/util/src/tsocket.c

@@ -270,6 +270,14 @@ int taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clientI
     return -1;
   }
 
+  /* set REUSEADDR option, so the portnumber can be re-used */
+  int reuse = 1;
+  if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(reuse)) < 0) {
+    uError("setsockopt SO_REUSEADDR failed: %d (%s)", errno, strerror(errno));
+    close(sockFd);
+    return -1;
+  };
+
   if ( clientIp != 0) {
     memset((char *)&clientAddr, 0, sizeof(clientAddr));
     clientAddr.sin_family = AF_INET;

src/vnode/src/vnodeMain.c

@@ -299,10 +299,6 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   }
 #endif
 
-  // start continuous query
-  if (pVnode->role == TAOS_SYNC_ROLE_MASTER)
-    cqStart(pVnode->cq);
-
   pVnode->qMgmt = qOpenQueryMgmt(pVnode->vgId);
   pVnode->events = NULL;
   pVnode->status = TAOS_VN_STATUS_READY;
@@ -313,15 +309,6 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t vnodeStartStream(int32_t vnode) {
-  SVnodeObj* pVnode = vnodeAcquire(vnode);
-  if (pVnode != NULL) {
-    tsdbStartStream(pVnode->tsdb);
-    vnodeRelease(pVnode);
-  }
-  return TSDB_CODE_SUCCESS;
-}
-
 int32_t vnodeClose(int32_t vgId) {
   SVnodeObj **ppVnode = (SVnodeObj **)taosHashGet(tsDnodeVnodesHash, (const char *)&vgId, sizeof(int32_t));
   if (ppVnode == NULL || *ppVnode == NULL) return 0;
@@ -354,11 +341,6 @@ void vnodeRelease(void *pVnodeRaw) {
     tsdbCloseRepo(pVnode->tsdb, 1);
   pVnode->tsdb = NULL;
 
-  // stop continuous query
-  if (pVnode->cq) 
-    cqClose(pVnode->cq);
-  pVnode->cq = NULL;
-
   if (pVnode->wal) 
     walClose(pVnode->wal);
   pVnode->wal = NULL;
@@ -525,8 +507,16 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
 
   // stop replication module
   if (pVnode->sync) {
-    syncStop(pVnode->sync);
+    void *sync = pVnode->sync;
     pVnode->sync = NULL;
+    syncStop(sync);
+  }
+
+  // stop continuous query
+  if (pVnode->cq) {
+    void *cq = pVnode->cq;
+    pVnode->cq = NULL;
+    cqClose(cq);
   }
 
   vTrace("vgId:%d, vnode will cleanup, refCount:%d", pVnode->vgId, pVnode->refCount);

tests/pytest/fulltest.sh

@@ -143,6 +143,7 @@ python3 ./test.py -f query/filterOtherTypes.py
 python3 ./test.py -f query/querySort.py
 python3 ./test.py -f query/queryJoin.py
 python3 ./test.py -f query/select_last_crash.py
+python3 ./test.py -f query/queryNullValueTest.py
 
 #stream
 python3 ./test.py -f stream/metric_1.py

tests/pytest/query/queryInsertValue.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+from util.dnodes import *
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+        
+        self.numOfRecords = 10
+        self.ts = 1537146000000    
+    
+    def restartTaosd(self):
+        tdDnodes.stop(1)
+        tdDnodes.start(1)
+        tdSql.execute("use db")
+
+    def run(self):
+        tdSql.prepare()
+
+        print("==============step1")
+
+        tdSql.execute(
+            "create table st (ts timestamp, speed int) tags(areaid int, loc nchar(20))")
+        tdSql.execute("create table t1 using st tags(1, 'beijing')")
+        tdSql.execute("insert into t1 values(now, 1)")
+        tdSql.query("select * from st")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table st add column length int")
+        tdSql.execute("insert into t1 values(now, 1, 2)")
+        tdSql.query("select last(*) from st")
+        tdSql.checkData(0, 2, 2);
+
+        self.restartTaosd();
+
+        tdSql.query("select last(*) from st")
+        tdSql.checkData(0, 2, 2);
+
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/query/queryNullValueTest.py

+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+import numpy as np
+from util.dnodes import *
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+        
+        self.numOfRecords = 10
+        self.ts = 1537146000000
+
+    def checkNullValue(self, result):
+        mx = np.array(result)
+        [rows, cols] = mx.shape
+        for i in range(rows):
+            for j in range(cols):
+                if j + 1 < cols and mx[i, j + 1] is not None:
+                    print(mx[i, j + 1])
+                    return False
+        return True
+    
+    def restartTaosd(self):
+        tdDnodes.stop(1)
+        tdDnodes.start(1)
+        tdSql.execute("use db")
+
+    def run(self):
+        tdSql.prepare()
+
+        print("==============step1")
+
+        tdSql.execute(
+            "create table meters (ts timestamp, col1 int) tags(tgcol1 int)")
+        tdSql.execute("create table t0 using meters tags(NULL)")
+
+        for i in range (self.numOfRecords):
+            tdSql.execute("insert into t0 values (%d, %d)" % (self.ts + i, i));
+
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col2 tinyint")
+        tdSql.execute("alter table meters drop column col1")        
+        tdSql.query("select * from meters")        
+        tdSql.checkRows(10)
+        tdSql.query("select col2 from meters")        
+        tdSql.checkRows(10)        
+
+        tdSql.execute("alter table meters add column col1 int")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col1 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col3 smallint")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col3 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col4 bigint")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col4 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col5 float")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col5 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col6 double")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col6 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col7 bool")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col7 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col8 binary(20)")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col8 from meters")        
+        tdSql.checkRows(10)
+
+        tdSql.execute("alter table meters add column col9 nchar(20)")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select col9 from meters")        
+        tdSql.checkRows(10)
+        
+        tdSql.execute("alter table meters add tag tgcol2 tinyint")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol2 from meters")
+        tdSql.checkRows(1)
+                
+
+        tdSql.execute("alter table meters add tag tgcol3 smallint")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol3 from meters")
+        tdSql.checkRows(1)
+        
+
+        tdSql.execute("alter table meters add tag tgcol4 bigint")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol4 from meters")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table meters add tag tgcol5 float")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol5 from meters")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table meters add tag tgcol6 double")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol6 from meters")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table meters add tag tgcol7 bool")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol7 from meters")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table meters add tag tgcol8 binary(20)")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol8 from meters")
+        tdSql.checkRows(1)
+
+        tdSql.execute("alter table meters add tag tgcol9 nchar(20)")
+        tdSql.query("select * from meters")
+        tdSql.checkRows(10)
+        tdSql.query("select tgcol9 from meters")
+        tdSql.checkRows(1)
+
+        self.restartTaosd()
+        tdSql.query("select * from meters")        
+        tdSql.checkRows(10)
+        if self.checkNullValue(tdSql.queryResult) is False:
+            tdLog.exit("non None value is detected")
+
+
+        
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/regressiontest.sh

@@ -140,6 +140,7 @@ python3 ./test.py -f query/queryJoin.py
 python3 ./test.py -f query/filterCombo.py
 python3 ./test.py -f query/queryNormal.py
 python3 ./test.py -f query/select_last_crash.py
+python3 ./test.py -f query/queryNullValueTest.py
 
 #stream
 python3 ./test.py -f stream/stream1.py