diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index d7f1881209a78ee6f00c3e0bb2528f5eb6bf4bdd..5356a71579df57f00fab6690e2c07f9940f9d3b6 100644
--- a/src/client/inc/tsclient.h
+++ b/src/client/inc/tsclient.h
@@ -404,6 +404,7 @@ TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port,
                      void *param, void **taos);
 void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ;
 
+int doAsyncParseSql(SSqlObj* pSql);
 void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
 
 void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql);
diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c
index 3414ff8d98f1854e5b97efc027fe14fca86cd81b..a9c170f1977a6d3da72d85cd76484a64d7286602 100644
--- a/src/client/src/tscAsync.c
+++ b/src/client/src/tscAsync.c
@@ -40,39 +40,38 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
 static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
 static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
 
-void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
-  SSqlCmd *pCmd = &pSql->cmd;
-  SSqlRes *pRes = &pSql->res;
+int doAsyncParseSql(SSqlObj* pSql) {
+  SSqlCmd* pCmd = &pSql->cmd;
+  SSqlRes* pRes = &pSql->res;
+  int32_t code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
+  if (code != TSDB_CODE_SUCCESS) {
+    tscError("failed to malloc payload");
+    tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY);
+    return code;
+  }
   
+  pRes->qhandle = 0;
+  pRes->numOfRows = 1;
+  
+  tscDump("%p SQL: %s", pSql, pSql->sqlstr);
+  return tsParseSql(pSql, true);
+}
+
+void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
   pSql->signature = pSql;
   pSql->param     = param;
   pSql->pTscObj   = pObj;
   pSql->maxRetry  = TSDB_MAX_REPLICA_NUM;
   pSql->fp        = fp;
-  
-  sem_init(&pSql->rspSem, 0, 0);
-  if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
-    tscError("failed to malloc payload");
-    tscQueueAsyncError(fp, param, TSDB_CODE_TSC_OUT_OF_MEMORY);
-    return;
-  }
-  
-  // todo check for OOM problem
   pSql->sqlstr = calloc(1, sqlLen + 1);
   if (pSql->sqlstr == NULL) {
     tscError("%p failed to malloc sql string buffer", pSql);
-    tscQueueAsyncError(fp, param, TSDB_CODE_TSC_OUT_OF_MEMORY);
-    free(pCmd->payload);
+    tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY);
     return;
   }
-  
-  pRes->qhandle = 0;
-  pRes->numOfRows = 1;
-  
   strtolower(pSql->sqlstr, sqlstr);
-  tscDump("%p SQL: %s", pSql, pSql->sqlstr);
   
-  int32_t code = tsParseSql(pSql, true);
+  int32_t code = doAsyncParseSql(pSql);
   if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return;
   
   if (code != TSDB_CODE_SUCCESS) {
@@ -518,15 +517,11 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
 
   if (pSql->pStream) {
     tscTrace("%p stream:%p meta is updated, start new query, command:%d", pSql, pSql->pStream, pSql->cmd.command);
-    /*
-     * NOTE:
-     * transfer the sql function for super table query before get meter/metric meta,
-     * since in callback functions, only tscProcessSql(pStream->pSql) is executed!
-     */
-    SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
-    
-    tscTansformSQLFuncForSTableQuery(pQueryInfo);
-    tscIncStreamExecutionCount(pSql->pStream);
+    if (!pSql->cmd.parseFinished) {
+      tsParseSql(pSql, false);
+      sem_post(&pSql->rspSem);
+    }
+    return;
   } else {
     tscTrace("%p get tableMeta successfully", pSql);
   }
diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c
index 15706e3235eb8f0c1dd80a8602574cac372369a1..f46ebe96f05f88fb30a610ea32936947f84e67bf 100644
--- a/src/client/src/tscSQLParser.c
+++ b/src/client/src/tscSQLParser.c
@@ -515,8 +515,9 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
         if (ret != 0) {
           return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
         }
-      }
+      } 
 
+      pCmd->parseFinished = 1;
       return TSDB_CODE_SUCCESS;  // do not build query message here
     }
 
diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 4e92e9dd70d1cb9ec6a98e6ca6ca676f6e95684d..c4413f85414f67d8db3cb5422acc5e34e5eb87c3 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -19,6 +19,7 @@
 #include "tscLog.h"
 #include "tscUtil.h"
 #include "tsched.h"
+#include "tcache.h"
 #include "tsclient.h"
 #include "ttime.h"
 #include "ttimer.h"
@@ -77,30 +78,23 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
   int code = tscGetTableMeta(pSql, pTableMetaInfo);
   pSql->res.code = code;
 
-  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return;
-
   if (code == 0 && UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
     code = tscGetSTableVgroupInfo(pSql, 0);
     pSql->res.code = code;
-
-    if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return;
   }
 
-  tscTansformSQLFuncForSTableQuery(pQueryInfo);
-
   // failed to get meter/metric meta, retry in 10sec.
   if (code != TSDB_CODE_SUCCESS) {
     int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
     tscError("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
-  
     tscSetRetryTimer(pStream, pSql, retryDelayTime);
-    return;
-  }
-
-  tscTrace("%p stream:%p start stream query on:%s", pSql, pStream, pTableMetaInfo->name);
-  tscProcessSql(pStream->pSql);
 
-  tscIncStreamExecutionCount(pStream);
+  } else {
+    tscTansformSQLFuncForSTableQuery(pQueryInfo);
+    tscTrace("%p stream:%p start stream query on:%s", pSql, pStream, pTableMetaInfo->name);
+    tscDoQuery(pStream->pSql);
+    tscIncStreamExecutionCount(pStream);
+  }
 }
 
 static void tscProcessStreamTimer(void *handle, void *tmrId) {
@@ -147,7 +141,8 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
              retryDelay);
 
     STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pStream->pSql->cmd, 0, 0);
-    tscClearTableMetaInfo(pTableMetaInfo, true);
+    taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pTableMeta), true);
+    tfree(pTableMetaInfo->vgroupList);
   
     tscSetRetryTimer(pStream, pStream->pSql, retryDelay);
     return;
@@ -259,7 +254,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
              pStream->numOfRes);
 
     // release the metric/meter meta information reference, so data in cache can be updated
-    tscClearTableMetaInfo(pTableMetaInfo, false);
+
+    taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pTableMeta), false);
+    tfree(pTableMetaInfo->vgroupList);
     tscSetNextLaunchTimer(pStream, pSql);
   }
 }
@@ -480,45 +477,37 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
 
   SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
   if (pSql == NULL) {
-    setErrorInfo(pSql, TSDB_CODE_TSC_OUT_OF_MEMORY, NULL);
     return NULL;
   }
 
   pSql->signature = pSql;
+  pSql->param = pSql;
   pSql->pTscObj = pObj;
+
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
-  int ret = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
-  if (TSDB_CODE_SUCCESS != ret) {
-    setErrorInfo(pSql, ret, NULL);
-    free(pSql);
+
+  SSqlStream *pStream = (SSqlStream *)calloc(1, sizeof(SSqlStream));
+  if (pStream == NULL) {
+    tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
+    tscFreeSqlObj(pSql);
     return NULL;
   }
+  pSql->pStream = pStream;
 
-  pSql->sqlstr = strdup(sqlstr);
+  pSql->sqlstr = calloc(1, strlen(sqlstr) + 1);
   if (pSql->sqlstr == NULL) {
-    setErrorInfo(pSql, TSDB_CODE_TSC_OUT_OF_MEMORY, NULL);
-
-    tfree(pSql);
-    return NULL;
+    tscError("%p failed to malloc sql string buffer", pSql);
+    tscFreeSqlObj(pSql);
+    return NULL;;
   }
+  strtolower(pSql->sqlstr, sqlstr);
 
   tsem_init(&pSql->rspSem, 0, 0);
-
-  SSqlInfo SQLInfo = {0};
-  tSQLParse(&SQLInfo, pSql->sqlstr);
-
-  tscResetSqlCmdObj(&pSql->cmd);
-  ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
-  if (TSDB_CODE_SUCCESS != ret) {
-    setErrorInfo(pSql, ret, NULL);
-    tscError("%p open stream failed, sql:%s, code:%d", pSql, sqlstr, TSDB_CODE_TSC_OUT_OF_MEMORY);
-    tscFreeSqlObj(pSql);
-    return NULL;
+  int32_t code = doAsyncParseSql(pSql);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    sem_wait(&pSql->rspSem);
   }
-  
-  pRes->code = tscToSQLCmd(pSql, &SQLInfo);
-  SQLInfoDestroy(&SQLInfo);
 
   if (pRes->code != TSDB_CODE_SUCCESS) {
     setErrorInfo(pSql, pRes->code, pCmd->payload);
@@ -528,15 +517,6 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
     return NULL;
   }
 
-  SSqlStream *pStream = (SSqlStream *)calloc(1, sizeof(SSqlStream));
-  if (pStream == NULL) {
-    setErrorInfo(pSql, TSDB_CODE_TSC_OUT_OF_MEMORY, NULL);
-
-    tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
-    tscFreeSqlObj(pSql);
-    return NULL;
-  }
-
   SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
   STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
@@ -550,13 +530,13 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   pStream->ctime = taosGetTimestamp(pStream->precision);
   pStream->etime = pQueryInfo->window.ekey;
 
-  pSql->pStream = pStream;
   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);
 
   tscTrace("%p stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql,
diff --git a/src/common/inc/tdataformat.h b/src/common/inc/tdataformat.h
index 3a84ce8f6f3ad369e9f6972d79d3356451c4ef25..ec4e544e18c44653ea1013759cb2db1ea26fb100 100644
--- a/src/common/inc/tdataformat.h
+++ b/src/common/inc/tdataformat.h
@@ -145,6 +145,7 @@ void     tdFreeDataRow(SDataRow row);
 void     tdInitDataRow(SDataRow row, STSchema *pSchema);
 SDataRow tdDataRowDup(SDataRow row);
 
+// offset here not include dataRow header length
 static FORCE_INLINE int tdAppendColVal(SDataRow row, void *value, int8_t type, int32_t bytes, int32_t offset) {
   ASSERT(value != NULL);
   int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c
index 9406a2fdce90713d784758d479e953aa18aacdb9..edb588b5544531d538529e3e027cf3dedeb44a0d 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -38,6 +38,7 @@ typedef struct {
   int      vgId;
   char     user[TSDB_USER_LEN];
   char     pass[TSDB_PASSWORD_LEN];
+  char     db[TSDB_DB_NAME_LEN];
   FCqWrite cqWrite;
   void    *ahandle;
   int      num;      // number of continuous streams
@@ -48,7 +49,8 @@ typedef struct {
 } SCqContext;
 
 typedef struct SCqObj {
-  int            tid;      // table ID
+  uint64_t       uid;
+  int32_t        tid;      // table ID
   int            rowSize;  // bytes of a row
   char *         sqlStr;   // SQL string
   STSchema *     pSchema;  // pointer to schema array
@@ -73,6 +75,14 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
 
   strcpy(pContext->user, pCfg->user);
   strcpy(pContext->pass, pCfg->pass);
+  const char* db = pCfg->db;
+  for (const char* p = db; *p != 0; p++) {
+    if (*p == '.') {
+      db = p + 1;
+      break;
+    }
+  }
+  strcpy(pContext->db, db);
   pContext->vgId = pCfg->vgId;
   pContext->cqWrite = pCfg->cqWrite;
   pContext->ahandle = ahandle;
@@ -153,17 +163,19 @@ void cqStop(void *handle) {
   pthread_mutex_unlock(&pContext->mutex);
 }
 
-void *cqCreate(void *handle, int tid, char *sqlStr, STSchema *pSchema) {
+void *cqCreate(void *handle, uint64_t uid, int tid, char *sqlStr, STSchema *pSchema) {
   SCqContext *pContext = handle;
 
   SCqObj *pObj = calloc(sizeof(SCqObj), 1);
   if (pObj == NULL) return NULL;
 
+  pObj->uid = uid;
   pObj->tid = tid;
   pObj->sqlStr = malloc(strlen(sqlStr)+1);
   strcpy(pObj->sqlStr, sqlStr);
 
   pObj->pSchema = tdDupSchema(pSchema);
+  pObj->rowSize = schemaTLen(pSchema);
 
   cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
 
@@ -207,16 +219,16 @@ void cqDrop(void *handle) {
 }
 
 static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
-
   if (pContext->dbConn == NULL) {
-    pContext->dbConn = taos_connect("localhost", pContext->user, pContext->pass, NULL, 0);
+    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;
     }
-    return;
   }
 
   int64_t lastKey = 0;
+  pObj->pContext = pContext;
   pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, lastKey, pObj, NULL);
   if (pObj->pStream) {
     pContext->num++;
@@ -229,29 +241,49 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
 static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   SCqObj     *pObj = (SCqObj *)param;
   SCqContext *pContext = pObj->pContext;
+  STSchema   *pSchema = pObj->pSchema;
   if (pObj->pStream == NULL) return;
 
   cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
 
-  // construct data
-  int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + pObj->rowSize;
+  int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize;
   char *buffer = calloc(size, 1);
 
   SWalHead   *pHead = (SWalHead *)buffer;
-  pHead->msgType = TSDB_MSG_TYPE_SUBMIT;
-  pHead->len = size - sizeof(SWalHead);
-  
-  SSubmitMsg *pSubmit = (SSubmitMsg *) (buffer + sizeof(SWalHead));
-  // to do: fill in the SSubmitMsg structure
-  pSubmit->numOfBlocks = 1;
+  SSubmitMsg *pMsg = (SSubmitMsg *) (buffer + sizeof(SWalHead));
+  SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
 
+  SDataRow trow = (SDataRow)pBlk->data;
+  tdInitDataRow(trow, pSchema);
 
-  SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
-  // to do: fill in the SSubmitBlk strucuture
-  pBlk->tid = pObj->tid;
+  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
+    STColumn *c = pSchema->columns + i;
+    char* val = (char*)row[i];
+    if (IS_VAR_DATA_TYPE(c->type)) {
+      val -= sizeof(VarDataLenT);
+    }
+    tdAppendColVal(trow, val, c->type, c->bytes, c->offset);
+  }
+  pBlk->len = htonl(dataRowLen(trow));
+
+  pBlk->uid = htobe64(pObj->uid);
+  pBlk->tid = htonl(pObj->tid);
+  pBlk->numOfRows = htons(1);
+  pBlk->sversion = htonl(pSchema->version);
+  pBlk->padding = 0;
 
+  pHead->len = sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + dataRowLen(trow);
+
+  pMsg->header.vgId = htonl(pContext->vgId);
+  pMsg->header.contLen = htonl(pHead->len);
+  pMsg->length = pMsg->header.contLen;
+  pMsg->numOfBlocks = htonl(1);
+
+  pHead->msgType = TSDB_MSG_TYPE_SUBMIT;
+  pHead->version = 0;
 
   // write into vnode write queue
   pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ);
+  free(buffer);
 }
 
diff --git a/src/cq/test/cqtest.c b/src/cq/test/cqtest.c
index fbe3c95b8642c45dea67daef2fa6fdaa702da314..1416a591bec5e4ec61e7a4581db955e91feec2e8 100644
--- a/src/cq/test/cqtest.c
+++ b/src/cq/test/cqtest.c
@@ -70,7 +70,7 @@ int main(int argc, char *argv[]) {
   tdDestroyTSchemaBuilder(&schemaBuilder);
 
   for (int sid =1; sid<10; ++sid) {
-    cqCreate(pCq, sid, "select avg(speed) from demo.t1 sliding(1s) interval(5s)", pSchema);
+    cqCreate(pCq, sid, sid, "select avg(speed) from demo.t1 sliding(1s) interval(5s)", pSchema);
   }
 
   tdFreeSchema(pSchema);
diff --git a/src/dnode/inc/dnodeModule.h b/src/dnode/inc/dnodeModule.h
index 8618de324446ca3f2504d15a6422bcca3a4b51b0..6a6da0a2a52844e4a2c42bad64658c93b48fffae 100644
--- a/src/dnode/inc/dnodeModule.h
+++ b/src/dnode/inc/dnodeModule.h
@@ -22,6 +22,7 @@ extern "C" {
 
 int32_t dnodeInitModules();
 void    dnodeStartModules();
+void    dnodeStartStream();
 void    dnodeCleanupModules();
 void    dnodeProcessModuleStatus(uint32_t moduleStatus);
 
diff --git a/src/dnode/src/dnodeMain.c b/src/dnode/src/dnodeMain.c
index 7683843371ab4ba0a0bfb6d3b119fbbeca031b8d..73bc2923b25f6d0d4b433a6247cc531a1d98325f 100644
--- a/src/dnode/src/dnodeMain.c
+++ b/src/dnode/src/dnodeMain.c
@@ -124,6 +124,7 @@ int32_t dnodeInitSystem() {
 
   dnodeStartModules();
   dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING);
+  dnodeStartStream();
 
   dPrint("TDengine is initialized successfully");
 
diff --git a/src/dnode/src/dnodeMgmt.c b/src/dnode/src/dnodeMgmt.c
index 69e91fd4ee41dc13c552523e253a6ff8a999224f..4b448837fe9c9f0f4dd8a997b54b8011096f9657 100644
--- a/src/dnode/src/dnodeMgmt.c
+++ b/src/dnode/src/dnodeMgmt.c
@@ -260,11 +260,27 @@ static int32_t dnodeOpenVnodes() {
   }
 
   free(vnodeList);
-
   dPrint("there are total vnodes:%d, openned:%d failed:%d", numOfVnodes, numOfVnodes-failed, failed);
   return TSDB_CODE_SUCCESS;
 }
 
+void dnodeStartStream() {
+  int32_t vnodeList[TSDB_MAX_VNODES];
+  int32_t numOfVnodes = 0;
+  int32_t status = dnodeGetVnodeList(vnodeList, &numOfVnodes);
+
+  if (status != TSDB_CODE_SUCCESS) {
+    dPrint("Get dnode list failed");
+    return;
+  }
+
+  for (int32_t i = 0; i < numOfVnodes; ++i) {
+    vnodeStartStream(vnodeList[i]);
+  }
+
+  dPrint("streams started");
+}
+
 static void dnodeCloseVnodes() {
   int32_t *vnodeList = (int32_t *)malloc(sizeof(int32_t) * TSDB_MAX_VNODES);
   int32_t numOfVnodes;
diff --git a/src/inc/tcq.h b/src/inc/tcq.h
index e025afaa0ae428b9545b5a23308830f8aa455002..9d987da468eaa83a0324d106fa19563c95608e6d 100644
--- a/src/inc/tcq.h
+++ b/src/inc/tcq.h
@@ -27,6 +27,7 @@ typedef struct {
   int      vgId;
   char     user[TSDB_USER_LEN];
   char     pass[TSDB_PASSWORD_LEN];
+  char     db[TSDB_DB_NAME_LEN + 1];
   FCqWrite cqWrite;
 } SCqCfg;
 
@@ -41,7 +42,7 @@ void  cqStart(void *handle);
 void  cqStop(void *handle);
 
 // cqCreate is called by TSDB to start an instance of CQ 
-void *cqCreate(void *handle, int sid, char *sqlStr, STSchema *pSchema);
+void *cqCreate(void *handle, uint64_t uid, int sid, char *sqlStr, STSchema *pSchema);
 
 // cqDrop is called by TSDB to stop an instance of CQ, handle is the return value of cqCreate
 void  cqDrop(void *handle);
diff --git a/src/inc/tsdb.h b/src/inc/tsdb.h
index 2dc9b977b46449602381606ade0508035ab01a26..a678f213bb5318980a56aa8a7bfb5c466e717c5c 100644
--- a/src/inc/tsdb.h
+++ b/src/inc/tsdb.h
@@ -43,7 +43,7 @@ typedef struct {
   void *cqH;
   int (*notifyStatus)(void *, int status);
   int (*eventCallBack)(void *);
-  void *(*cqCreateFunc)(void *handle, int sid, char *sqlStr, STSchema *pSchema);
+  void *(*cqCreateFunc)(void *handle, uint64_t uid, int sid, char *sqlStr, STSchema *pSchema);
   void (*cqDropFunc)(void *handle);
   void *(*configFunc)(int32_t vgId, int32_t sid);
 } STsdbAppH;
@@ -118,6 +118,7 @@ int   tsdbDropTable(TsdbRepoT *pRepo, STableId tableId);
 int   tsdbAlterTable(TsdbRepoT *repo, STableCfg *pCfg);
 int   tsdbUpdateTagValue(TsdbRepoT *repo, SUpdateTableTagValMsg *pMsg);
 TSKEY tsdbGetTableLastKey(TsdbRepoT *repo, uint64_t uid);
+void  tsdbStartStream(TsdbRepoT *repo);
 
 uint32_t tsdbGetFileInfo(TsdbRepoT *repo, char *name, uint32_t *index, uint32_t eindex, int32_t *size);
 
diff --git a/src/inc/vnode.h b/src/inc/vnode.h
index f4fb8060feabc12e0f6bfcb933572ed4090ebb6e..e507c55942bc6b4db682fd25739a9da34ce668fb 100644
--- a/src/inc/vnode.h
+++ b/src/inc/vnode.h
@@ -38,6 +38,7 @@ 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);
 
diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c
index 8f102795b5bf07888a09438b8f1a13bf2d03b231..5e32a9e7d76a2767203a887ec10383f2a28ce80a 100644
--- a/src/tsdb/src/tsdbMain.c
+++ b/src/tsdb/src/tsdbMain.c
@@ -473,6 +473,18 @@ TSKEY tsdbGetTableLastKey(TsdbRepoT *repo, uint64_t uid) {
   return TSDB_GET_TABLE_LAST_KEY(pTable);
 }
 
+void  tsdbStartStream(TsdbRepoT *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, pTable->tableId.uid, pTable->tableId.tid, pTable->sql, tsdbGetTableSchema(pMeta, pTable));
+    }
+  }
+}
+
 STableInfo *tsdbGetTableInfo(TsdbRepoT *pRepo, STableId tableId) {
   // TODO
   return NULL;
diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c
index 34d3c2789365ff10a6d95a65e347cafe2cb1bd78..05c3b303773bd2542f9c35e535fb6309f5eb58ce 100644
--- a/src/tsdb/src/tsdbMeta.c
+++ b/src/tsdb/src/tsdbMeta.c
@@ -150,7 +150,6 @@ int tsdbRestoreTable(void *pHandle, void *cont, int contLen) {
 
 void tsdbOrgMeta(void *pHandle) {
   STsdbMeta *pMeta = (STsdbMeta *)pHandle;
-  STsdbRepo *pRepo = (STsdbRepo *)pMeta->pRepo;
 
   for (int i = 1; i < pMeta->maxTables; i++) {
     STable *pTable = pMeta->tables[i];
@@ -158,13 +157,6 @@ void tsdbOrgMeta(void *pHandle) {
       tsdbAddTableIntoIndex(pMeta, pTable);
     }
   }
-
-  for (int i = 0; i < pMeta->maxTables; i++) {
-    STable *pTable = pMeta->tables[i];
-    if (pTable && pTable->type == TSDB_STREAM_TABLE) {
-      pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, i, pTable->sql, tsdbGetTableSchema(pMeta, pTable));
-    }
-  }
 }
 
 /**
@@ -683,7 +675,7 @@ static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx) {
       tsdbAddTableIntoIndex(pMeta, pTable);
     }
     if (pTable->type == TSDB_STREAM_TABLE && addIdx) {
-      pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, pTable->tableId.tid, pTable->sql, tsdbGetTableSchema(pMeta, pTable));
+      pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, pTable->tableId.uid, pTable->tableId.tid, pTable->sql, tsdbGetTableSchema(pMeta, pTable));
     }
     
     pMeta->nTables++;
diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c
index d1dfa24cbeaa692d64602c4f0ee319c8953d316b..256ef3c72baf0ccc5dd0c39b1f6452a9dfd204d6 100644
--- a/src/vnode/src/vnodeMain.c
+++ b/src/vnode/src/vnodeMain.c
@@ -208,8 +208,9 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   }
 
   SCqCfg cqCfg = {0};
-  sprintf(cqCfg.user, "root");
+  sprintf(cqCfg.user, "_root");
   strcpy(cqCfg.pass, tsInternalPass);
+  strcpy(cqCfg.db, pVnode->db);
   cqCfg.vgId = vnode;
   cqCfg.cqWrite = vnodeWriteToQueue;
   pVnode->cq = cqOpen(pVnode, &cqCfg);
@@ -277,6 +278,15 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t vnodeStartStream(int32_t vnode) {
+  SVnodeObj* pVnode = vnodeAccquireVnode(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;
diff --git a/tests/pytest/stream/stream1.py b/tests/pytest/stream/stream1.py
new file mode 100644
index 0000000000000000000000000000000000000000..7a9d88da3b2aa7bedae0824cac89349dcbeb5040
--- /dev/null
+++ b/tests/pytest/stream/stream1.py
@@ -0,0 +1,131 @@
+###################################################################
+#           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 time
+import taos
+from util.log import tdLog
+from util.cases import tdCases
+from util.sql import tdSql
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+
+    def run(self):
+        tbNum = 10
+        rowNum = 20
+
+        tdSql.prepare()
+
+        tdLog.info("===== step1 =====")
+        tdSql.execute("create table stb0(ts timestamp, col1 int, col2 float) tags(tgcol int)")
+        for i in range(tbNum):
+            tdSql.execute("create table tb%d using stb0 tags(%d)" % (i, i))
+            for j in range(rowNum):
+                tdSql.execute("insert into tb%d values (now - %dm, %d, %d)" % (i, 1440 - j, j, j))
+        time.sleep(0.1)
+
+        tdLog.info("===== step2 =====")
+        tdSql.query("select count(*), count(col1), count(col2) from tb0 interval(1d)")
+        tdSql.checkData(0, 1, rowNum)
+        tdSql.checkData(0, 2, rowNum)
+        tdSql.checkData(0, 3, rowNum)
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum)
+        tdSql.execute("create table s0 as select count(*), count(col1), count(col2) from tb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step3 =====")
+        tdLog.info("sleeping 120 seconds")
+        time.sleep(120)
+        tdSql.query("select * from s0")
+        tdSql.checkData(0, 1, rowNum)
+        tdSql.checkData(0, 2, rowNum)
+        tdSql.checkData(0, 3, rowNum)
+
+        tdLog.info("===== step4 =====")
+        tdSql.execute("drop table s0")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum)
+
+        tdLog.info("===== step5 =====")
+        tdSql.error("select * from s0")
+
+        tdLog.info("===== step6 =====")
+        time.sleep(0.1)
+        tdSql.execute("create table s0 as select count(*), count(col1), count(col2) from tb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step7 =====")
+        tdLog.info("sleeping 120 seconds")
+        time.sleep(120)
+
+        tdSql.query("select * from s0")
+        tdSql.checkData(0, 1, rowNum)
+        tdSql.checkData(0, 2, rowNum)
+        tdSql.checkData(0, 3, rowNum)
+
+        tdLog.info("===== step8 =====")
+        tdSql.query("select count(*), count(col1), count(col2) from stb0 interval(1d)")
+        tdSql.checkData(0, 1, rowNum * tbNum)
+        tdSql.checkData(0, 2, rowNum * tbNum)
+        tdSql.checkData(0, 3, rowNum * tbNum)
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdSql.execute("create table s1 as select count(*), count(col1), count(col2) from stb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 2)
+
+        tdLog.info("===== step9 =====")
+        tdLog.info("sleeping 120 seconds")
+        time.sleep(120)
+
+        tdSql.query("select * from s1")
+        tdSql.checkData(0, 1, rowNum * tbNum)
+        tdSql.checkData(0, 2, rowNum * tbNum)
+        tdSql.checkData(0, 3, rowNum * tbNum)
+
+        tdLog.info("===== step10 =====")
+        tdSql.execute("drop table s1")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step11 =====")
+        tdSql.error("select * from s1")
+
+        tdLog.info("===== step12 =====")
+        tdSql.execute("create table s1 as select count(*), count(col1), count(col2) from stb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 2)
+
+        tdLog.info("===== step13 =====")
+        tdLog.info("sleeping 120 seconds")
+        time.sleep(120)
+        tdSql.query("select * from s1")
+        tdSql.checkData(0, 1, rowNum * tbNum)
+        tdSql.checkData(0, 2, rowNum * tbNum)
+        tdSql.checkData(0, 3, rowNum * tbNum)
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
diff --git a/tests/pytest/stream/stream2.py b/tests/pytest/stream/stream2.py
new file mode 100644
index 0000000000000000000000000000000000000000..96eff3131d5bec3eb7e16f7d002aae25da560bf2
--- /dev/null
+++ b/tests/pytest/stream/stream2.py
@@ -0,0 +1,122 @@
+###################################################################
+#           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 time
+import taos
+from util.log import tdLog
+from util.cases import tdCases
+from util.sql import tdSql
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), logSql)
+
+
+    def run(self):
+        tbNum = 10
+        rowNum = 20
+        totalNum = tbNum * rowNum
+
+        tdSql.prepare()
+
+        tdLog.info("===== step1 =====")
+        tdSql.execute("create table stb0(ts timestamp, col1 int, col2 float) tags(tgcol int)")
+        for i in range(tbNum):
+            tdSql.execute("create table tb%d using stb0 tags(%d)" % (i, i))
+            for j in range(rowNum):
+                tdSql.execute("insert into tb%d values (now - %dm, %d, %d)" % (i, 1440 - j, j, j))
+        time.sleep(0.1)
+
+        tdLog.info("===== step2 =====")
+        tdSql.query("select count(col1) from tb0 interval(1d)")
+        tdSql.checkData(0, 1, rowNum)
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum)
+        tdSql.execute("create table s0 as select count(col1) from tb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step3 =====")
+        time.sleep(120)
+        tdSql.query("select * from s0")
+        tdSql.checkData(0, 1, rowNum)
+
+        tdLog.info("===== step4 =====")
+        tdSql.execute("drop table s0")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum)
+
+        tdLog.info("===== step5 =====")
+        tdSql.error("select * from s0")
+
+        tdLog.info("===== step6 =====")
+        tdSql.execute("create table s0 as select count(*), count(col1), count(col2) from tb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step7 =====")
+        time.sleep(120)
+        tdSql.query("select * from s0")
+        tdSql.checkData(0, 1, rowNum)
+        tdSql.checkData(0, 2, rowNum)
+        tdSql.checkData(0, 3, rowNum)
+
+        tdLog.info("===== step8 =====")
+        tdSql.query("select count(*), count(col1), count(col2) from stb0 interval(1d)")
+        tdSql.checkData(0, 1, totalNum)
+        tdSql.checkData(0, 2, totalNum)
+        tdSql.checkData(0, 3, totalNum)
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+        tdSql.execute("create table s1 as select count(*), count(col1), count(col2) from stb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 2)
+
+        tdLog.info("===== step9 =====")
+        time.sleep(120)
+        tdSql.query("select * from s1")
+        tdSql.checkData(0, 1, totalNum)
+        tdSql.checkData(0, 2, totalNum)
+        tdSql.checkData(0, 3, totalNum)
+
+        tdLog.info("===== step10 =====")
+        tdSql.execute("drop table s1")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 1)
+
+        tdLog.info("===== step11 =====")
+        tdSql.error("select * from s1")
+
+        tdLog.info("===== step12 =====")
+        tdSql.execute("create table s1 as select count(col1) from stb0 interval(1d)")
+        tdSql.query("show tables")
+        tdSql.checkRows(tbNum + 2)
+
+        tdLog.info("===== step13 =====")
+        time.sleep(120)
+        tdSql.query("select * from s1")
+        tdSql.checkData(0, 1, totalNum)
+        #tdSql.checkData(0, 2, None)
+        #tdSql.checkData(0, 3, None)
+
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
diff --git a/tests/script/general/stream/metrics_1.sim b/tests/script/general/stream/metrics_1.sim
index c60bde6b2f00c6b888899faef75c21a2c732db9e..94498cb92591966b6d289c440d675c93fa752005 100644
--- a/tests/script/general/stream/metrics_1.sim
+++ b/tests/script/general/stream/metrics_1.sim
@@ -205,8 +205,8 @@ if $data01 != 20 then
 endi
 
 print =============== step21
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step22
 $st = $stPrefix . c1
diff --git a/tests/script/general/stream/metrics_del.sim b/tests/script/general/stream/metrics_del.sim
index c856871e93b7be436cd7da6d665cb90439c9ae7f..e21fa5999aa1b6ce0c063af54172f222aeb983e1 100644
--- a/tests/script/general/stream/metrics_del.sim
+++ b/tests/script/general/stream/metrics_del.sim
@@ -76,20 +76,20 @@ endw
 sql drop table $mt
 
 print =============== step4
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step5
 $st = $stPrefix . c3
 sql select * from $st
 print ===> select * from $st
 print ===> $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
-if $data01 != NULL then 
+if $data01 != null then 
   return -1
 endi
-if $data02 != NULL then 
+if $data02 != null then 
   return -1
 endi
-if $data03 != NULL then 
+if $data03 != null then 
   return -1
 endi
diff --git a/tests/script/general/stream/metrics_n.sim b/tests/script/general/stream/metrics_n.sim
index 47089403dd2f64c3cb5b216ed8258433d74c40d6..7fc08064b29cc6f1681116be16ca460da0e41580 100644
--- a/tests/script/general/stream/metrics_n.sim
+++ b/tests/script/general/stream/metrics_n.sim
@@ -187,8 +187,8 @@ $st = $stPrefix . as
 #sql create table $st as select avg(tbcol) as a1, sum(tbcol) as a2, min(tbcol) as a3, max(tbcol) as a4,  first(tbcol) as a5, last(tbcol) as a6, count(tbcol) as a7, avg(tbcol) as a8, sum(tbcol) as a9, min(tbcol) as a3, max(tbcol) as a4,  first(tbcol) as a5, last(tbcol) as a6, count(tbcol) as a7 from $mt where ts < now + 4m interval(1d)
 
 print =============== step9
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step10
 $st = $stPrefix . c3
diff --git a/tests/script/general/stream/metrics_replica1_vnoden.sim b/tests/script/general/stream/metrics_replica1_vnoden.sim
index 13cd18adf312e01781482ed82918c47b7ce3e1cb..ee071fd681ae117ec6ff0b7b2846a197a7950d97 100644
--- a/tests/script/general/stream/metrics_replica1_vnoden.sim
+++ b/tests/script/general/stream/metrics_replica1_vnoden.sim
@@ -163,8 +163,8 @@ $st = $stPrefix . as
 sql create table $st as select count(tbcol) as c from $mt interval(1d)
 
 print =============== step13
-print sleep 22 seconds
-sleep 32000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step14
 $st = $stPrefix . c1
diff --git a/tests/script/general/stream/new_stream.sim b/tests/script/general/stream/new_stream.sim
index 8aa0a898705d67590212db7cec0e81ceecb3d8f2..abc06faaef7418e1d86450831853c9e8ad65e434 100644
--- a/tests/script/general/stream/new_stream.sim
+++ b/tests/script/general/stream/new_stream.sim
@@ -1,12 +1,12 @@
-system sh/stop_dnodes.sh
+#system sh/stop_dnodes.sh
 
 
-system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c walLevel -v 0
-system sh/cfg.sh -n dnode1 -c tableMetaKeepTimer -v 10
-system sh/exec.sh -n dnode1 -s start
+#system sh/deploy.sh -n dnode1 -i 1
+#system sh/cfg.sh -n dnode1 -c walLevel -v 0
+#system sh/cfg.sh -n dnode1 -c tableMetaKeepTimer -v 10
+#system sh/exec.sh -n dnode1 -s start
 
-sleep 3000
+#sleep 3000
 sql connect
 print ======================== dnode1 start
 
@@ -56,14 +56,14 @@ print $data00 $data01 $data02 $data03
 sql create table $st as select count(*), count(tbcol), count(tbcol2) from $mt interval(10s)
 
 print =============== step3
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step4
 
 sql select * from $st
 print $st ==> $rows1 $data00 $data01 $data02 $data03
-if $data13 >= 51 then
+if $data03 >= 51 then
  return -1
 endi
 
@@ -90,8 +90,8 @@ while $i < $tbNum
 endw 
 
 print =============== step6
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step7
 
diff --git a/tests/script/general/stream/restart_stream.sim b/tests/script/general/stream/restart_stream.sim
index aa16934f330c2cf7a1e8e67f1738f737ef846eb5..138e2a6e2ef6e6aeb46c14b7e8fceb851e36eaa8 100644
--- a/tests/script/general/stream/restart_stream.sim
+++ b/tests/script/general/stream/restart_stream.sim
@@ -73,8 +73,8 @@ print =============== step3
 sql create table $stt as select count(*) from $tb interval(1d)
 sql create table $stm as select count(*) from $mt interval(1d)
 
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 sql select * from $stt
 print select count(*) from $stt ===> $data00 $data01
@@ -152,8 +152,8 @@ print =============== step8
 sql create table $stt as select count(*) from $tb interval(1d)
 sql create table $stm as select count(*) from $mt interval(1d)
 
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 sql select * from $stt
 sleep 1000
diff --git a/tests/script/general/stream/stream_1.sim b/tests/script/general/stream/stream_1.sim
index 2309d341c1464cd455055691a68a5fefad03b762..958c877ee57725edaf18e5c586d96c98540bfa79 100644
--- a/tests/script/general/stream/stream_1.sim
+++ b/tests/script/general/stream/stream_1.sim
@@ -78,8 +78,8 @@ if $rows != 11 then
 endi
 
 print =============== step3
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 20 then 
@@ -112,8 +112,8 @@ if $rows != 11 then
 endi
 
 print =============== step7
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 20 then 
@@ -155,8 +155,8 @@ if $rows != 12 then
 endi
 
 print =============== step9
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 200 then 
@@ -190,8 +190,8 @@ if $rows != 12 then
 endi
 
 print =============== step13
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 200 then 
diff --git a/tests/script/general/stream/stream_2.sim b/tests/script/general/stream/stream_2.sim
index 13aac983374b40258f12f6a4b5d7e7e7dd143a0b..057529b427b5e76dd778fd6e8cbca9504a52e1af 100644
--- a/tests/script/general/stream/stream_2.sim
+++ b/tests/script/general/stream/stream_2.sim
@@ -72,8 +72,8 @@ if $rows != 11 then
 endi
 
 print =============== step3
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 20 then 
@@ -100,8 +100,8 @@ if $rows != 11 then
 endi
 
 print =============== step7
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st => $data01
 if $data01 != 20 then 
@@ -143,8 +143,8 @@ if $rows != 12 then
 endi
 
 print =============== step9
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st =>  $data01 $data02, $data03
 if $data01 != 200 then 
@@ -178,17 +178,17 @@ if $rows != 12 then
 endi
 
 print =============== step13
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 sql select * from $st
 print select * from $st =>  $data01 $data02, $data03
 if $data01 != 200 then 
   return -1
 endi
-if $data02 != NULL then 
+if $data02 != null then 
   return -1
 endi
-if $data03 != NULL then 
+if $data03 != null then 
   return -1
 endi
 
diff --git a/tests/script/general/stream/stream_3.sim b/tests/script/general/stream/stream_3.sim
index 914ecd848492dd837024417c2341d135cad25c1d..88105a77d667e9cad499ec55cad7f0d7f8fc0c3f 100644
--- a/tests/script/general/stream/stream_3.sim
+++ b/tests/script/general/stream/stream_3.sim
@@ -79,8 +79,8 @@ $st = $stPrefix . c3
 sql create table $st as select count(tbcol2) from $tb interval(1d)
 
 print =============== step5
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step6
 $st = $stPrefix . c1
@@ -173,8 +173,8 @@ $st = $stPrefix . c3
 sql create table $st as select count(*), count(tbcol), count(tbcol2) from $tb interval(1d)
 
 print =============== step10
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step11
 #$st = $stPrefix . c3
diff --git a/tests/script/general/stream/stream_restart.sim b/tests/script/general/stream/stream_restart.sim
index d5fbef49083d266d38c8702990aaec61039252c9..480b23055e90f59bd597d3319a5fe139eba83bab 100644
--- a/tests/script/general/stream/stream_restart.sim
+++ b/tests/script/general/stream/stream_restart.sim
@@ -79,8 +79,8 @@ sleep 1000
 system sh/exec.sh -n dnode1 -s start
 
 print =============== step4
-print sleep 23 seconds
-sleep 23000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step5
 $i = 1
diff --git a/tests/script/general/stream/table_1.sim b/tests/script/general/stream/table_1.sim
index efc7c370cab780919442fdf705e26bd273b15364..f028b1626d156a65a3ff805f1479e9cb178a8d30 100644
--- a/tests/script/general/stream/table_1.sim
+++ b/tests/script/general/stream/table_1.sim
@@ -214,8 +214,8 @@ sql select count(tbcol) from $tb where ts < now + 4m interval(1d) group by tgcol
 step20:
 
 print =============== step21
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step22
 $st = $stPrefix . c1
diff --git a/tests/script/general/stream/table_del.sim b/tests/script/general/stream/table_del.sim
index 66fd58308f06cd053e063cde8b77b1ec18515169..ce4065a1a85ba2c4f8f79b77d1d82c5118091e9c 100644
--- a/tests/script/general/stream/table_del.sim
+++ b/tests/script/general/stream/table_del.sim
@@ -71,20 +71,20 @@ print =============== step3
 sql drop table $tb
 
 print =============== step4
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step5
 $st = $stPrefix . c3
 sql select * from $st
 print ===> select * from $st
 print ===> $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
-if $data01 != NULL then 
+if $data01 != null then 
   return -1
 endi
-if $data02 != NULL then 
+if $data02 != null then 
   return -1
 endi
-if $data03 != NULL then 
+if $data03 != null then 
   return -1
 endi
diff --git a/tests/script/general/stream/table_n.sim b/tests/script/general/stream/table_n.sim
index e6037c5292b9e0da7f5fbd57a67fda88795a26ee..d1b4a87a9ea641a8f4d58e0d9fc61bd9ac7c7477 100644
--- a/tests/script/general/stream/table_n.sim
+++ b/tests/script/general/stream/table_n.sim
@@ -191,8 +191,8 @@ $st = $stPrefix . as
 #sql create table $st as select avg(tbcol) as a1, sum(tbcol) as a2, min(tbcol) as a3, max(tbcol) as a4,  first(tbcol) as a5, last(tbcol) as a6, stddev(tbcol) as a7, percentile(tbcol, 1) as a8, count(tbcol) as a9, leastsquares(tbcol, 1, 1) as a10 from $tb where ts < now + 4m interval(1d)
 
 print =============== step10
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step11
 $st = $stPrefix . c3
diff --git a/tests/script/general/stream/table_replica1_vnoden.sim b/tests/script/general/stream/table_replica1_vnoden.sim
index 44d4008dbd36594575472477b4a36887bc5f25ba..e1d5a9babf5147ed9d4815ba00608d116d5b3773 100644
--- a/tests/script/general/stream/table_replica1_vnoden.sim
+++ b/tests/script/general/stream/table_replica1_vnoden.sim
@@ -196,8 +196,8 @@ $st = $stPrefix . as
 sql create table $st as select count(tbcol) as c from $tb interval(1d)
 
 print =============== step16
-print sleep 22 seconds
-sleep 22000 
+print sleep 120 seconds
+sleep 120000 
 
 print =============== step17
 $st = $stPrefix . c1