From d34cabf2c1a782fe7001e3f82699aadfd6f3911c Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 29 May 2020 17:15:09 +0800
Subject: [PATCH 1/7] TD-449: fix several bugs

---
 src/cq/src/cqMain.c         |  6 ++++--
 src/dnode/inc/dnodeModule.h |  3 ++-
 src/dnode/src/dnodeMain.c   |  1 +
 src/dnode/src/dnodeMgmt.c   | 18 +++++++++++++++++-
 src/inc/tcq.h               |  1 +
 src/inc/tsdb.h              |  1 +
 src/inc/vnode.h             |  1 +
 src/tsdb/src/tsdbMain.c     | 12 ++++++++++++
 src/tsdb/src/tsdbMeta.c     |  8 --------
 src/vnode/src/vnodeMain.c   | 12 +++++++++++-
 10 files changed, 50 insertions(+), 13 deletions(-)

diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c
index 9406a2fdce..799a0b0514 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
@@ -73,6 +74,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
 
   strcpy(pContext->user, pCfg->user);
   strcpy(pContext->pass, pCfg->pass);
+  strcpy(pContext->db, pCfg->db);
   pContext->vgId = pCfg->vgId;
   pContext->cqWrite = pCfg->cqWrite;
   pContext->ahandle = ahandle;
@@ -207,9 +209,8 @@ 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));
     }
@@ -217,6 +218,7 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
   }
 
   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++;
diff --git a/src/dnode/inc/dnodeModule.h b/src/dnode/inc/dnodeModule.h
index 8618de3244..fb529ee67c 100644
--- a/src/dnode/inc/dnodeModule.h
+++ b/src/dnode/inc/dnodeModule.h
@@ -22,7 +22,8 @@ extern "C" {
 
 int32_t dnodeInitModules();
 void    dnodeStartModules();
-void    dnodeCleanupModules();
+void    dnodeStartStream();
+void    dnodeCleanUpModules();
 void    dnodeProcessModuleStatus(uint32_t moduleStatus);
 
 #ifdef __cplusplus
diff --git a/src/dnode/src/dnodeMain.c b/src/dnode/src/dnodeMain.c
index 7683843371..73bc2923b2 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 7c457defca..a29f99dda6 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 e025afaa0a..ba198ab66e 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];
   FCqWrite cqWrite;
 } SCqCfg;
 
diff --git a/src/inc/tsdb.h b/src/inc/tsdb.h
index 2dc9b977b4..d3ec96acf6 100644
--- a/src/inc/tsdb.h
+++ b/src/inc/tsdb.h
@@ -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 f4fb8060fe..e507c55942 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 0efb477e73..632c1dd34e 100644
--- a/src/tsdb/src/tsdbMain.c
+++ b/src/tsdb/src/tsdbMain.c
@@ -474,6 +474,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.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 9f1e507d54..bf5fac4e45 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));
-    }
-  }
 }
 
 /**
diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c
index 3d8cf67d02..037dadb8fd 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, "s1_db0");   // TODO: replace hard coded db name
   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;

From 69142ea5aaa29af8fca4dea9115d88665abcd307 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Mon, 1 Jun 2020 17:17:22 +0800
Subject: [PATCH 2/7] TD-449: data is written to TSDB

---
 src/client/src/tscStream.c   |  8 ++++--
 src/common/inc/tdataformat.h |  1 +
 src/cq/src/cqMain.c          | 50 ++++++++++++++++++++++++++----------
 src/cq/test/cqtest.c         |  2 +-
 src/inc/tcq.h                |  2 +-
 src/inc/tsdb.h               |  2 +-
 src/tsdb/src/tsdbMain.c      |  2 +-
 src/tsdb/src/tsdbMeta.c      |  2 +-
 src/vnode/src/vnodeMain.c    |  2 +-
 9 files changed, 50 insertions(+), 21 deletions(-)

diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 9e1628bb9b..3c7cd3e1eb 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"
@@ -147,7 +148,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 +261,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);
   }
 }
diff --git a/src/common/inc/tdataformat.h b/src/common/inc/tdataformat.h
index 3a84ce8f6f..e706f9a9d5 100644
--- a/src/common/inc/tdataformat.h
+++ b/src/common/inc/tdataformat.h
@@ -70,6 +70,7 @@ typedef struct {
   int      numOfCols;  // Number of columns appended
   int      tlen;       // maximum length of a SDataRow without the header part
   int      flen;       // First part length in a SDataRow after the header part
+  int32_t  version;
   STColumn columns[];
 } STSchema;
 
diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c
index 799a0b0514..d96e0fe6a7 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -49,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
@@ -155,17 +156,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 = pSchema->tlen;
 
   cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
 
@@ -213,8 +216,8 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
     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;
@@ -231,6 +234,7 @@ 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);
@@ -240,18 +244,38 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   char *buffer = calloc(size, 1);
 
   SWalHead   *pHead = (SWalHead *)buffer;
+  SSubmitMsg *pMsg = (SSubmitMsg *) (buffer + sizeof(SWalHead));
+  SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
+
+  int32_t flen = 0;
+  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
+    flen += TYPE_BYTES[pSchema->columns[i].type];
+  }
+
+  SDataRow trow = (SDataRow)pBlk->data;
+  dataRowSetLen(trow, TD_DATA_ROW_HEAD_SIZE + flen);
+
+  int toffset = 0;
+  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
+    tdAppendColVal(trow, row[i], pSchema->columns[i].type, pSchema->columns[i].bytes, toffset);
+    toffset += TYPE_BYTES[pSchema->columns[i].type];
+  }
+  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;
+
+  pMsg->header.vgId = htonl(pContext->vgId);
+  pMsg->header.contLen = htonl(size - sizeof(SWalHead));
+  pMsg->length = pMsg->header.contLen;
+  pMsg->numOfBlocks = htonl(1);
+
   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;
-
-
-  SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
-  // to do: fill in the SSubmitBlk strucuture
-  pBlk->tid = pObj->tid;
-
+  pHead->version = 0;
 
   // write into vnode write queue
   pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ);
diff --git a/src/cq/test/cqtest.c b/src/cq/test/cqtest.c
index fbe3c95b86..1416a591be 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/inc/tcq.h b/src/inc/tcq.h
index ba198ab66e..32b75674c3 100644
--- a/src/inc/tcq.h
+++ b/src/inc/tcq.h
@@ -42,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 d3ec96acf6..a678f213bb 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;
diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c
index 632c1dd34e..f8fea779f3 100644
--- a/src/tsdb/src/tsdbMain.c
+++ b/src/tsdb/src/tsdbMain.c
@@ -481,7 +481,7 @@ void  tsdbStartStream(TsdbRepoT *repo) {
   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.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));
     }
   }
 }
diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c
index bf5fac4e45..ac0f6b26ac 100644
--- a/src/tsdb/src/tsdbMeta.c
+++ b/src/tsdb/src/tsdbMeta.c
@@ -675,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 037dadb8fd..26ba392a76 100644
--- a/src/vnode/src/vnodeMain.c
+++ b/src/vnode/src/vnodeMain.c
@@ -210,7 +210,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   SCqCfg cqCfg = {0};
   sprintf(cqCfg.user, "_root");
   strcpy(cqCfg.pass, tsInternalPass);
-  strcpy(cqCfg.db, "s1_db0");   // TODO: replace hard coded db name
+  strcpy(cqCfg.db, "db");   // TODO: replace hard coded db name
   cqCfg.vgId = vnode;
   cqCfg.cqWrite = vnodeWriteToQueue;
   pVnode->cq = cqOpen(pVnode, &cqCfg);

From 40c1d665b5481850c4a0e0604675f5eedf648e03 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Tue, 2 Jun 2020 18:12:33 +0800
Subject: [PATCH 3/7] fix some issues

---
 src/client/src/tscStream.c   | 14 ++++++++++++++
 src/common/inc/tdataformat.h |  1 -
 src/cq/src/cqMain.c          | 13 +++++++------
 src/dnode/inc/dnodeModule.h  |  2 +-
 src/inc/tcq.h                |  2 +-
 src/vnode/src/vnodeMain.c    |  2 +-
 6 files changed, 24 insertions(+), 10 deletions(-)

diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 3c7cd3e1eb..83a097f2f5 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -477,6 +477,14 @@ static void setErrorInfo(SSqlObj* pSql, int32_t code, char* info) {
   }
 }
 
+static void asyncCallback(void *param, TAOS_RES *tres, int code) {
+  assert(param != NULL);
+  SSqlObj *pSql = ((SSqlObj *)param);
+  
+  pSql->res.code = code;
+  sem_post(&pSql->rspSem);
+}
+
 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;
@@ -521,7 +529,13 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
     return NULL;
   }
   
+  pSql->param = pSql;
+  pSql->fp = asyncCallback;
   pRes->code = tscToSQLCmd(pSql, &SQLInfo);
+  if (pRes->code == TSDB_CODE_ACTION_IN_PROGRESS) {
+    sem_wait(&pSql->rspSem);
+  }
+  
   SQLInfoDestroy(&SQLInfo);
 
   if (pRes->code != TSDB_CODE_SUCCESS) {
diff --git a/src/common/inc/tdataformat.h b/src/common/inc/tdataformat.h
index e706f9a9d5..3a84ce8f6f 100644
--- a/src/common/inc/tdataformat.h
+++ b/src/common/inc/tdataformat.h
@@ -70,7 +70,6 @@ typedef struct {
   int      numOfCols;  // Number of columns appended
   int      tlen;       // maximum length of a SDataRow without the header part
   int      flen;       // First part length in a SDataRow after the header part
-  int32_t  version;
   STColumn columns[];
 } STSchema;
 
diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c
index d96e0fe6a7..401d61b7a2 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -239,19 +239,19 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
 
   cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
 
+  int32_t flen = 0;
+  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
+    flen += TYPE_BYTES[pSchema->columns[i].type];
+  }
+
   // 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 + flen;
   char *buffer = calloc(size, 1);
 
   SWalHead   *pHead = (SWalHead *)buffer;
   SSubmitMsg *pMsg = (SSubmitMsg *) (buffer + sizeof(SWalHead));
   SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
 
-  int32_t flen = 0;
-  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
-    flen += TYPE_BYTES[pSchema->columns[i].type];
-  }
-
   SDataRow trow = (SDataRow)pBlk->data;
   dataRowSetLen(trow, TD_DATA_ROW_HEAD_SIZE + flen);
 
@@ -279,5 +279,6 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
 
   // write into vnode write queue
   pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ);
+  free(buffer);
 }
 
diff --git a/src/dnode/inc/dnodeModule.h b/src/dnode/inc/dnodeModule.h
index fb529ee67c..6a6da0a2a5 100644
--- a/src/dnode/inc/dnodeModule.h
+++ b/src/dnode/inc/dnodeModule.h
@@ -23,7 +23,7 @@ extern "C" {
 int32_t dnodeInitModules();
 void    dnodeStartModules();
 void    dnodeStartStream();
-void    dnodeCleanUpModules();
+void    dnodeCleanupModules();
 void    dnodeProcessModuleStatus(uint32_t moduleStatus);
 
 #ifdef __cplusplus
diff --git a/src/inc/tcq.h b/src/inc/tcq.h
index 32b75674c3..9d987da468 100644
--- a/src/inc/tcq.h
+++ b/src/inc/tcq.h
@@ -27,7 +27,7 @@ typedef struct {
   int      vgId;
   char     user[TSDB_USER_LEN];
   char     pass[TSDB_PASSWORD_LEN];
-  char     db[TSDB_DB_NAME_LEN];
+  char     db[TSDB_DB_NAME_LEN + 1];
   FCqWrite cqWrite;
 } SCqCfg;
 
diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c
index 26ba392a76..e4f1f7d2f5 100644
--- a/src/vnode/src/vnodeMain.c
+++ b/src/vnode/src/vnodeMain.c
@@ -210,7 +210,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   SCqCfg cqCfg = {0};
   sprintf(cqCfg.user, "_root");
   strcpy(cqCfg.pass, tsInternalPass);
-  strcpy(cqCfg.db, "db");   // TODO: replace hard coded db name
+  strcpy(cqCfg.db, pVnode->db);
   cqCfg.vgId = vnode;
   cqCfg.cqWrite = vnodeWriteToQueue;
   pVnode->cq = cqOpen(pVnode, &cqCfg);

From 81ea4a8a8617d03d301935529ee2e7c7d9ff9bbb Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Wed, 3 Jun 2020 16:16:41 +0800
Subject: [PATCH 4/7] first test case passed

---
 src/client/inc/tsclient.h      |   1 +
 src/client/src/tscAsync.c      |  37 +++++-----
 src/client/src/tscStream.c     |  39 ++--------
 src/cq/src/cqMain.c            |   9 ++-
 tests/pytest/stream/stream1.py | 131 +++++++++++++++++++++++++++++++++
 5 files changed, 166 insertions(+), 51 deletions(-)
 create mode 100644 tests/pytest/stream/stream1.py

diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index d7f1881209..e92bb5e64e 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, const char* sqlstr, size_t sqlLen);
 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 ebcdddffde..bd2dc64ef0 100644
--- a/src/client/src/tscAsync.c
+++ b/src/client/src/tscAsync.c
@@ -40,30 +40,23 @@ 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;
-  
-  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)) {
+int doAsyncParseSql(SSqlObj* pSql, const char* sqlstr, size_t sqlLen) {
+  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(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
-    return;
+    tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
+    return code;
   }
   
   // 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_CLI_OUT_OF_MEMORY);
+    tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
     free(pCmd->payload);
-    return;
+    return TSDB_CODE_CLI_OUT_OF_MEMORY;
   }
   
   pRes->qhandle = 0;
@@ -72,7 +65,17 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
   strtolower(pSql->sqlstr, sqlstr);
   tscDump("%p SQL: %s", pSql, pSql->sqlstr);
   
-  int32_t code = tsParseSql(pSql, true);
+  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;
+  
+  int32_t code = doAsyncParseSql(pSql, sqlstr, sqlLen);
   if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
   
   if (code != TSDB_CODE_SUCCESS) {
diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 83a097f2f5..028ae0592a 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -497,46 +497,18 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   }
 
   pSql->signature = pSql;
+  pSql->param = pSql;
   pSql->pTscObj = pObj;
+  pSql->fp = asyncCallback;
+
   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);
-    return NULL;
-  }
-
-  pSql->sqlstr = strdup(sqlstr);
-  if (pSql->sqlstr == NULL) {
-    setErrorInfo(pSql, TSDB_CODE_CLI_OUT_OF_MEMORY, NULL);
-
-    tfree(pSql);
-    return NULL;
-  }
 
   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_CLI_OUT_OF_MEMORY);
-    tscFreeSqlObj(pSql);
-    return NULL;
-  }
-  
-  pSql->param = pSql;
-  pSql->fp = asyncCallback;
-  pRes->code = tscToSQLCmd(pSql, &SQLInfo);
-  if (pRes->code == TSDB_CODE_ACTION_IN_PROGRESS) {
+  int32_t code = doAsyncParseSql(pSql, sqlstr, strlen(sqlstr));
+  if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
     sem_wait(&pSql->rspSem);
   }
-  
-  SQLInfoDestroy(&SQLInfo);
 
   if (pRes->code != TSDB_CODE_SUCCESS) {
     setErrorInfo(pSql, pRes->code, pCmd->payload);
@@ -575,6 +547,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   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/cq/src/cqMain.c b/src/cq/src/cqMain.c
index 401d61b7a2..5c936388dd 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -75,7 +75,14 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
 
   strcpy(pContext->user, pCfg->user);
   strcpy(pContext->pass, pCfg->pass);
-  strcpy(pContext->db, pCfg->db);
+  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;
diff --git a/tests/pytest/stream/stream1.py b/tests/pytest/stream/stream1.py
new file mode 100644
index 0000000000..7a9d88da3b
--- /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())

From 51de20d2e0c0dea8bec77be77258c2db2283c328 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 5 Jun 2020 19:00:04 +0800
Subject: [PATCH 5/7] td-449: more cases passedcurrently, only below four cases
 fail:* agg_stream.sim ('us' ?)* column_stream.sim ('us' ?)*
 table_replica1_vnoden.sim (crash)* new_stream.sim (failed at line 101)

---
 src/client/inc/tsclient.h                     |   2 +-
 src/client/src/tscAsync.c                     |  34 ++---
 src/client/src/tscStream.c                    |  40 +++---
 src/common/inc/tdataformat.h                  |   1 +
 src/cq/src/cqMain.c                           |  19 ++-
 tests/pytest/stream/stream2.py                | 122 ++++++++++++++++++
 tests/script/general/stream/metrics_1.sim     |   4 +-
 tests/script/general/stream/metrics_del.sim   |  10 +-
 tests/script/general/stream/metrics_n.sim     |   4 +-
 .../stream/metrics_replica1_vnoden.sim        |   4 +-
 .../script/general/stream/restart_stream.sim  |   8 +-
 tests/script/general/stream/stream_1.sim      |  16 +--
 tests/script/general/stream/stream_2.sim      |  20 +--
 tests/script/general/stream/stream_3.sim      |   8 +-
 .../script/general/stream/stream_restart.sim  |   4 +-
 tests/script/general/stream/table_1.sim       |   4 +-
 tests/script/general/stream/table_del.sim     |  10 +-
 tests/script/general/stream/table_n.sim       |   4 +-
 18 files changed, 210 insertions(+), 104 deletions(-)
 create mode 100644 tests/pytest/stream/stream2.py

diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index e92bb5e64e..5356a71579 100644
--- a/src/client/inc/tsclient.h
+++ b/src/client/inc/tsclient.h
@@ -404,7 +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, const char* sqlstr, size_t sqlLen);
+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 bd2dc64ef0..3cf9e0daa5 100644
--- a/src/client/src/tscAsync.c
+++ b/src/client/src/tscAsync.c
@@ -40,7 +40,7 @@ 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);
 
-int doAsyncParseSql(SSqlObj* pSql, const char* sqlstr, size_t sqlLen) {
+int doAsyncParseSql(SSqlObj* pSql) {
   SSqlCmd* pCmd = &pSql->cmd;
   SSqlRes* pRes = &pSql->res;
   int32_t code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
@@ -50,21 +50,10 @@ int doAsyncParseSql(SSqlObj* pSql, const char* sqlstr, size_t sqlLen) {
     return code;
   }
   
-  // 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(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
-    free(pCmd->payload);
-    return TSDB_CODE_CLI_OUT_OF_MEMORY;
-  }
-  
   pRes->qhandle = 0;
   pRes->numOfRows = 1;
   
-  strtolower(pSql->sqlstr, sqlstr);
   tscDump("%p SQL: %s", pSql, pSql->sqlstr);
-  
   return tsParseSql(pSql, true);
 }
 
@@ -74,8 +63,15 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
   pSql->pTscObj   = pObj;
   pSql->maxRetry  = TSDB_MAX_REPLICA_NUM;
   pSql->fp        = fp;
+  pSql->sqlstr = calloc(1, sqlLen + 1);
+  if (pSql->sqlstr == NULL) {
+    tscError("%p failed to malloc sql string buffer", pSql);
+    tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
+    return;
+  }
+  strtolower(pSql->sqlstr, sqlstr);
   
-  int32_t code = doAsyncParseSql(pSql, sqlstr, sqlLen);
+  int32_t code = doAsyncParseSql(pSql);
   if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
   
   if (code != TSDB_CODE_SUCCESS) {
@@ -521,15 +517,9 @@ 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);
+    tsParseSql(pSql, false);
+    sem_post(&pSql->rspSem);
+    return;
   } else {
     tscTrace("%p get tableMeta successfully", pSql);
   }
diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 028ae0592a..c4c9860f3b 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -99,7 +99,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
   }
 
   tscTrace("%p stream:%p start stream query on:%s", pSql, pStream, pTableMetaInfo->name);
-  tscProcessSql(pStream->pSql);
+  tscDoQuery(pStream->pSql);
 
   tscIncStreamExecutionCount(pStream);
 }
@@ -477,14 +477,6 @@ static void setErrorInfo(SSqlObj* pSql, int32_t code, char* info) {
   }
 }
 
-static void asyncCallback(void *param, TAOS_RES *tres, int code) {
-  assert(param != NULL);
-  SSqlObj *pSql = ((SSqlObj *)param);
-  
-  pSql->res.code = code;
-  sem_post(&pSql->rspSem);
-}
-
 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;
@@ -492,20 +484,34 @@ 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_CLI_OUT_OF_MEMORY, NULL);
     return NULL;
   }
 
   pSql->signature = pSql;
   pSql->param = pSql;
   pSql->pTscObj = pObj;
-  pSql->fp = asyncCallback;
 
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
 
+  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 = calloc(1, strlen(sqlstr) + 1);
+  if (pSql->sqlstr == NULL) {
+    tscError("%p failed to malloc sql string buffer", pSql);
+    tscFreeSqlObj(pSql);
+    return NULL;;
+  }
+  strtolower(pSql->sqlstr, sqlstr);
+
   tsem_init(&pSql->rspSem, 0, 0);
-  int32_t code = doAsyncParseSql(pSql, sqlstr, strlen(sqlstr));
+  int32_t code = doAsyncParseSql(pSql);
   if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
     sem_wait(&pSql->rspSem);
   }
@@ -518,15 +524,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_CLI_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);
@@ -540,7 +537,6 @@ 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);
diff --git a/src/common/inc/tdataformat.h b/src/common/inc/tdataformat.h
index 3a84ce8f6f..ec4e544e18 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 5c936388dd..23455b6f50 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -246,13 +246,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
 
   cTrace("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
 
-  int32_t flen = 0;
-  for (int32_t i = 0; i < pSchema->numOfCols; i++) {
-    flen += TYPE_BYTES[pSchema->columns[i].type];
-  }
-
-  // construct data
-  int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + flen;
+  int size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize;
   char *buffer = calloc(size, 1);
 
   SWalHead   *pHead = (SWalHead *)buffer;
@@ -260,12 +254,15 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   SSubmitBlk *pBlk = (SSubmitBlk *) (buffer + sizeof(SWalHead) + sizeof(SSubmitMsg));
 
   SDataRow trow = (SDataRow)pBlk->data;
-  dataRowSetLen(trow, TD_DATA_ROW_HEAD_SIZE + flen);
+  tdInitDataRow(trow, pSchema);
 
-  int toffset = 0;
   for (int32_t i = 0; i < pSchema->numOfCols; i++) {
-    tdAppendColVal(trow, row[i], pSchema->columns[i].type, pSchema->columns[i].bytes, toffset);
-    toffset += TYPE_BYTES[pSchema->columns[i].type];
+    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));
 
diff --git a/tests/pytest/stream/stream2.py b/tests/pytest/stream/stream2.py
new file mode 100644
index 0000000000..96eff3131d
--- /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 c60bde6b2f..94498cb925 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 c856871e93..e21fa5999a 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 47089403dd..7fc08064b2 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 13cd18adf3..ee071fd681 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/restart_stream.sim b/tests/script/general/stream/restart_stream.sim
index aa16934f33..138e2a6e2e 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 2309d341c1..958c877ee5 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 13aac98337..057529b427 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 914ecd8484..88105a77d6 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 d5fbef4908..480b23055e 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 efc7c370ca..f028b1626d 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 66fd58308f..ce4065a1a8 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 e6037c5292..d1b4a87a9e 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

From 06ac5328ec801a07c18e38416e823814ef40934e Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Sat, 6 Jun 2020 16:55:31 +0800
Subject: [PATCH 6/7] td-449: fix invalid write

---
 src/cq/src/cqMain.c                                   | 7 ++++---
 tests/script/general/stream/table_replica1_vnoden.sim | 4 ++--
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c
index 23455b6f50..edb588b554 100644
--- a/src/cq/src/cqMain.c
+++ b/src/cq/src/cqMain.c
@@ -175,7 +175,7 @@ void *cqCreate(void *handle, uint64_t uid, int tid, char *sqlStr, STSchema *pSch
   strcpy(pObj->sqlStr, sqlStr);
 
   pObj->pSchema = tdDupSchema(pSchema);
-  pObj->rowSize = pSchema->tlen;
+  pObj->rowSize = schemaTLen(pSchema);
 
   cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr);
 
@@ -272,13 +272,14 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
   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(size - sizeof(SWalHead));
+  pMsg->header.contLen = htonl(pHead->len);
   pMsg->length = pMsg->header.contLen;
   pMsg->numOfBlocks = htonl(1);
 
   pHead->msgType = TSDB_MSG_TYPE_SUBMIT;
-  pHead->len = size - sizeof(SWalHead);
   pHead->version = 0;
 
   // write into vnode write queue
diff --git a/tests/script/general/stream/table_replica1_vnoden.sim b/tests/script/general/stream/table_replica1_vnoden.sim
index 44d4008dbd..e1d5a9babf 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

From 519cd6aee706726246c6f2cda8c560db92368aa1 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Mon, 8 Jun 2020 09:44:25 +0800
Subject: [PATCH 7/7] td-449: set parse finsish flag in select statement

---
 src/client/src/tscAsync.c                  |  6 ++++--
 src/client/src/tscSQLParser.c              |  3 ++-
 src/client/src/tscStream.c                 | 19 ++++++-------------
 tests/script/general/stream/new_stream.sim | 22 +++++++++++-----------
 4 files changed, 23 insertions(+), 27 deletions(-)

diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c
index 3723c0cf65..a9c170f197 100644
--- a/src/client/src/tscAsync.c
+++ b/src/client/src/tscAsync.c
@@ -517,8 +517,10 @@ 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);
-    tsParseSql(pSql, false);
-    sem_post(&pSql->rspSem);
+    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 68980bef98..1584b52e9f 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 f13e99d7b3..c4413f8541 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -78,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;
+
+  } else {
+    tscTansformSQLFuncForSTableQuery(pQueryInfo);
+    tscTrace("%p stream:%p start stream query on:%s", pSql, pStream, pTableMetaInfo->name);
+    tscDoQuery(pStream->pSql);
+    tscIncStreamExecutionCount(pStream);
   }
-
-  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) {
diff --git a/tests/script/general/stream/new_stream.sim b/tests/script/general/stream/new_stream.sim
index 8aa0a89870..abc06faaef 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