From c877418df00f36c1621a6ccc2e2adf65e0457611 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 00:31:23 +0000
Subject: [PATCH 01/25] support multi thread update mnode and dnode

---
 src/client/inc/tsclient.h  |  2 +-
 src/client/src/tscServer.c | 78 ++++++++++++++++++++++++++------------
 src/client/src/tscSql.c    |  2 +-
 src/client/src/tscSystem.c |  3 +-
 src/client/src/tscUtil.c   | 17 +++++----
 src/inc/trpc.h             |  5 +++
 6 files changed, 72 insertions(+), 35 deletions(-)

diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index be82eb64a8..e1180c4143 100644
--- a/src/client/inc/tsclient.h
+++ b/src/client/inc/tsclient.h
@@ -457,7 +457,7 @@ extern void *    tscQhandle;
 extern int       tscKeepConn[];
 extern int       tsInsertHeadSize;
 extern int       tscNumOfThreads;
-extern SRpcIpSet tscMgmtIpSet;
+extern SRpcCorIpSet tscMgmtIpSet;
 
 extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
 
diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index d8af6d5c87..e8a03a6482 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -29,7 +29,7 @@
 
 #define TSC_MGMT_VNODE 999
 
-SRpcIpSet  tscMgmtIpSet;
+SRpcCorIpSet  tscMgmtIpSet;
 SRpcIpSet  tscDnodeIpSet;
 
 int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo) = {0};
@@ -44,6 +44,42 @@ void tscSaveSubscriptionProgress(void* sub);
 
 static int32_t minMsgSize() { return tsRpcHeadSize + 100; }
 
+static void tscDumpMgmtIpSet(SRpcIpSet *ipSet) {
+  taosCorBeginRead(&tscMgmtIpSet.version);
+  *ipSet = tscMgmtIpSet.ipSet;
+  taosCorEndRead(&tscMgmtIpSet.version);
+}  
+
+bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
+   if (s1->numOfIps != s2->numOfIps 
+        || s1->inUse != s1->inUse) {
+     return false;
+   } 
+   for (int32_t i = 0; i < s1->numOfIps; i++) {
+     if (s1->port[i] != s2->port[i] 
+        || strncmp(s1->fqdn[i], s2->fqdn[i], TSDB_FQDN_LEN) != 0)
+        return false;
+   }
+   return true;
+}
+void tscSetMgmtIpList(SRpcIpSet *pIpSet) {
+  // no need to update if equal
+  SRpcIpSet dump;
+  tscDumpMgmtIpSet(&dump);
+  if (tscIpSetIsEqual(&dump, pIpSet)) {
+    return; 
+  }
+   
+  taosCorBeginWrite(&tscMgmtIpSet.version);
+  SRpcIpSet *mgmtIpSet = &tscMgmtIpSet.ipSet;
+  mgmtIpSet->numOfIps = pIpSet->numOfIps;
+  mgmtIpSet->inUse = pIpSet->inUse;
+  for (int32_t i = 0; i < mgmtIpSet->numOfIps; ++i) {
+     mgmtIpSet->port[i] = htons(pIpSet->port[i]);
+     strncpy(mgmtIpSet->fqdn[i], pIpSet->fqdn[i], TSDB_FQDN_LEN);
+  }
+  taosCorEndWrite(&tscMgmtIpSet.version);
+}
 static void tscSetDnodeIpList(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
   SRpcIpSet* pIpList = &pSql->ipList;
   pIpList->inUse    = 0;
@@ -60,31 +96,17 @@ static void tscSetDnodeIpList(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
 }
 
 void tscPrintMgmtIp() {
-  if (tscMgmtIpSet.numOfIps <= 0) {
-    tscError("invalid mnode IP list:%d", tscMgmtIpSet.numOfIps);
+  SRpcIpSet dump;
+  tscDumpMgmtIpSet(&dump);
+  if (dump.numOfIps <= 0) {
+    tscError("invalid mnode IP list:%d", dump.numOfIps);
   } else {
-    for (int i = 0; i < tscMgmtIpSet.numOfIps; ++i) {
-      tscDebug("mnode index:%d %s:%d", i, tscMgmtIpSet.fqdn[i], tscMgmtIpSet.port[i]);
+    for (int i = 0; i < dump.numOfIps; ++i) {
+      tscDebug("mnode index:%d %s:%d", i, dump.fqdn[i], dump.port[i]);
     }
   }
 }
 
-void tscSetMgmtIpList(SRpcIpSet *pIpList) {
-  tscMgmtIpSet.numOfIps = pIpList->numOfIps;
-  tscMgmtIpSet.inUse = pIpList->inUse;
-  for (int32_t i = 0; i < tscMgmtIpSet.numOfIps; ++i) {
-    tscMgmtIpSet.port[i] = htons(pIpList->port[i]);
-  }
-}
-
-void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
-  tscMgmtIpSet = *pIpSet;
-  tscDebug("mnode IP list is changed for ufp is called, numOfIps:%d inUse:%d", tscMgmtIpSet.numOfIps, tscMgmtIpSet.inUse);
-  for (int32_t i = 0; i < tscMgmtIpSet.numOfIps; ++i) {
-    tscDebug("index:%d fqdn:%s port:%d", i, tscMgmtIpSet.fqdn[i], tscMgmtIpSet.port[i]);
-  }
-}
-
 /*
  * For each management node, try twice at least in case of poor network situation.
  * If the client start to connect to a non-management node from the client, and the first retry may fail due to
@@ -95,7 +117,9 @@ void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet) {
 UNUSED_FUNC
 static int32_t tscGetMgmtConnMaxRetryTimes() {
   int32_t factor = 2;
-  return tscMgmtIpSet.numOfIps * factor;
+  SRpcIpSet dump;
+  tscDumpMgmtIpSet(&dump);
+  return dump.numOfIps * factor;
 }
 
 void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
@@ -185,7 +209,9 @@ int tscSendMsgToServer(SSqlObj *pSql) {
 
   // set the mgmt ip list
   if (pSql->cmd.command >= TSDB_SQL_MGMT) {
-    pSql->ipList = tscMgmtIpSet;
+    SRpcIpSet dump;
+    tscDumpMgmtIpSet(&dump);
+    pSql->ipList = dump;
   }
 
   memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
@@ -239,7 +265,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
   if (pCmd->command < TSDB_SQL_MGMT) {
     if (pIpSet) pSql->ipList = *pIpSet;
   } else {
-    if (pIpSet) tscMgmtIpSet = *pIpSet;
+    if (pIpSet) tscSetMgmtIpList(pIpSet);  
   }
 
   if (rpcMsg->pCont == NULL) {
@@ -421,7 +447,9 @@ int tscProcessSql(SSqlObj *pSql) {
       return pSql->res.code;
     }
   } else if (pCmd->command < TSDB_SQL_LOCAL) {
-    pSql->ipList = tscMgmtIpSet;
+    SRpcIpSet dump;
+    tscDumpMgmtIpSet(&dump);
+    pSql->ipList = dump;
   } else {  // local handler
     return (*tscProcessMsgRsp[pCmd->command])(pSql);
   }
diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c
index 0677463d8d..b24cfd9b27 100644
--- a/src/client/src/tscSql.c
+++ b/src/client/src/tscSql.c
@@ -63,7 +63,7 @@ SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
 
   if (ip) {
     if (tscSetMgmtIpListFromCfg(ip, NULL) < 0) return NULL;
-    if (port) tscMgmtIpSet.port[0] = port;
+    if (port) tscMgmtIpSet.ipSet.port[0] = port;
   } 
  
   void *pDnodeConn = NULL;
diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c
index 82cc8cc225..1a78ed87bd 100644
--- a/src/client/src/tscSystem.c
+++ b/src/client/src/tscSystem.c
@@ -41,7 +41,8 @@ int tscNumOfThreads;
 
 static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
 void taosInitNote(int numOfNoteLines, int maxNotes, char* lable);
-void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet);
+//void tscUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet);
+
 
 void tscCheckDiskUsage(void *UNUSED_PARAM(para), void* UNUSED_PARAM(param)) {
   taosGetDisk();
diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c
index 3e0fe0b4be..b0e6c727eb 100644
--- a/src/client/src/tscUtil.c
+++ b/src/client/src/tscUtil.c
@@ -2146,16 +2146,19 @@ char* strdup_throw(const char* str) {
 }
 
 int tscSetMgmtIpListFromCfg(const char *first, const char *second) {
-  tscMgmtIpSet.numOfIps = 0;
-  tscMgmtIpSet.inUse = 0;
+  // init mgmt ip set 
+  tscMgmtIpSet.version = 0;
+  SRpcIpSet *mgmtIpSet = &(tscMgmtIpSet.ipSet);
+  mgmtIpSet->numOfIps = 0;
+  mgmtIpSet->inUse = 0;
 
   if (first && first[0] != 0) {
     if (strlen(first) >= TSDB_EP_LEN) {
       terrno = TSDB_CODE_TSC_INVALID_FQDN;
       return -1;
     }
-    taosGetFqdnPortFromEp(first, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
-    tscMgmtIpSet.numOfIps++;
+    taosGetFqdnPortFromEp(first, mgmtIpSet->fqdn[mgmtIpSet->numOfIps], &(mgmtIpSet->port[mgmtIpSet->numOfIps]));
+    mgmtIpSet->numOfIps++;
   }
 
   if (second && second[0] != 0) {
@@ -2163,11 +2166,11 @@ int tscSetMgmtIpListFromCfg(const char *first, const char *second) {
       terrno = TSDB_CODE_TSC_INVALID_FQDN;
       return -1;
     }
-    taosGetFqdnPortFromEp(second, tscMgmtIpSet.fqdn[tscMgmtIpSet.numOfIps], &tscMgmtIpSet.port[tscMgmtIpSet.numOfIps]);
-    tscMgmtIpSet.numOfIps++;
+    taosGetFqdnPortFromEp(second, mgmtIpSet->fqdn[mgmtIpSet->numOfIps], &(mgmtIpSet->port[mgmtIpSet->numOfIps]));
+    mgmtIpSet->numOfIps++;
   }
 
-  if ( tscMgmtIpSet.numOfIps == 0) {
+  if (mgmtIpSet->numOfIps == 0) {
     terrno = TSDB_CODE_TSC_INVALID_FQDN;
     return -1;
   }
diff --git a/src/inc/trpc.h b/src/inc/trpc.h
index d1adfb7494..b159155e9d 100644
--- a/src/inc/trpc.h
+++ b/src/inc/trpc.h
@@ -35,6 +35,11 @@ typedef struct SRpcIpSet {
   char      fqdn[TSDB_MAX_REPLICA][TSDB_FQDN_LEN];
 } SRpcIpSet;
 
+typedef struct SRpcCorIpSet {
+  int32_t version; 
+  SRpcIpSet ipSet; 
+} SRpcCorIpSet;
+
 typedef struct SRpcConnInfo {
   uint32_t  clientIp;
   uint16_t  clientPort;

From 3d10b6f81d7d0f301bc3dd42405bf4f3869ab723 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 04:25:37 +0000
Subject: [PATCH 02/25] support multi thread update dnode

---
 src/client/src/tscSchemaUtil.c |  3 ++
 src/client/src/tscServer.c     | 67 +++++++++++++++++++++-------------
 src/inc/taosmsg.h              |  1 +
 3 files changed, 45 insertions(+), 26 deletions(-)

diff --git a/src/client/src/tscSchemaUtil.c b/src/client/src/tscSchemaUtil.c
index 934a562387..b391a5760e 100644
--- a/src/client/src/tscSchemaUtil.c
+++ b/src/client/src/tscSchemaUtil.c
@@ -157,6 +157,9 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
   pTableMeta->sid = pTableMetaMsg->sid;
   pTableMeta->uid = pTableMetaMsg->uid;
   pTableMeta->vgroupInfo = pTableMetaMsg->vgroup;
+  //init version here
+  pTableMeta->vgroupInfo.version = 0; 
+  
   pTableMeta->sversion = pTableMetaMsg->sversion;
   pTableMeta->tversion = pTableMetaMsg->tversion;
   
diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index e8a03a6482..4a11288a85 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -51,8 +51,7 @@ static void tscDumpMgmtIpSet(SRpcIpSet *ipSet) {
 }  
 
 bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
-   if (s1->numOfIps != s2->numOfIps 
-        || s1->inUse != s1->inUse) {
+   if (s1->numOfIps != s2->numOfIps /*|| s1->inUse != s1->inUse*/) {
      return false;
    } 
    for (int32_t i = 0; i < s1->numOfIps; i++) {
@@ -80,21 +79,38 @@ void tscSetMgmtIpList(SRpcIpSet *pIpSet) {
   }
   taosCorEndWrite(&tscMgmtIpSet.version);
 }
-static void tscSetDnodeIpList(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
-  SRpcIpSet* pIpList = &pSql->ipList;
-  pIpList->inUse    = 0;
-  if (pVgroupInfo == NULL) {
-    pIpList->numOfIps = 0;
+static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet, int32_t *vgId) {
+  if (pVgroupInfo == NULL) { return;}
+  taosCorBeginRead(&pVgroupInfo->version);
+  if (vgId) { 
+    *vgId = pVgroupInfo->vgId; 
+  }
+  pIpSet->inUse = 0; 
+  pIpSet->numOfIps = pVgroupInfo->numOfIps;  
+  for (int32_t i = 0; i < pVgroupInfo->numOfIps; ++i) {
+    strncpy(pIpSet->fqdn[i], pVgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
+    pIpSet->port[i] = pVgroupInfo->ipAddr[i].port;
+  }
+  taosCorEndRead(&pVgroupInfo->version);
+}
+static void tscSetVgroupInfoWithIpSet(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet) {
+  taosCorBeginWrite(&pVgroupInfo->version);
+  //TODO(dengyihao), dont care vgid  
+  pVgroupInfo->numOfIps = pIpSet->numOfIps;
+  for (int32_t i = 0; pVgroupInfo->numOfIps; i++) {
+    strncpy(pVgroupInfo->ipAddr[i].fqdn, pIpSet->fqdn[i], TSDB_FQDN_LEN);
+    pVgroupInfo->ipAddr[i].port = pIpSet->port[i];
+  }
+  taosCorEndWrite(&pVgroupInfo->version);
+}
+static void tscSetVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
+  if (tscIpSetIsEqual(&pObj->ipList, pIpSet)) {
     return;
   }
-  
-  pIpList->numOfIps = pVgroupInfo->numOfIps;
-  for(int32_t i = 0; i < pVgroupInfo->numOfIps; ++i) {
-    strcpy(pIpList->fqdn[i], pVgroupInfo->ipAddr[i].fqdn);
-    pIpList->port[i] = pVgroupInfo->ipAddr[i].port;
-  }
+  SSqlCmd *pCmd = &pObj->cmd;
+  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
+  tscSetVgroupInfoWithIpSet(&pTableMetaInfo->pTableMeta->vgroupInfo, pIpSet);  
 }
-
 void tscPrintMgmtIp() {
   SRpcIpSet dump;
   tscDumpMgmtIpSet(&dump);
@@ -263,7 +279,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
   }
 
   if (pCmd->command < TSDB_SQL_MGMT) {
-    if (pIpSet) pSql->ipList = *pIpSet;
+    if (pIpSet) tscSetVgroupInfo(pSql, pIpSet); 
   } else {
     if (pIpSet) tscSetMgmtIpList(pIpSet);  
   }
@@ -553,7 +569,7 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
 
   // pSql->cmd.payloadLen is set during copying data into payload
   pSql->cmd.msgType = TSDB_MSG_TYPE_SUBMIT;
-  tscSetDnodeIpList(pSql, &pTableMeta->vgroupInfo);
+  tscDumpIpSetFromVgroupInfo(&pTableMeta->vgroupInfo, &pSql->ipList, NULL);
   
   tscDebug("%p build submit msg, vgId:%d numOfTables:%d numberOfIP:%d", pSql, vgId, pSql->cmd.numOfTablesInSubmit,
       pSql->ipList.numOfIps);
@@ -595,11 +611,9 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     } else {
       pVgroupInfo = &pTableMeta->vgroupInfo;
     }
-
-    tscSetDnodeIpList(pSql, pVgroupInfo);
-    if (pVgroupInfo != NULL) {
-      pQueryMsg->head.vgId = htonl(pVgroupInfo->vgId);
-    }
+    int32_t vgId = 0;
+    tscDumpIpSetFromVgroupInfo(pVgroupInfo, &pSql->ipList, &vgId);
+    pQueryMsg->head.vgId = htonl(vgId);
 
     STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
     pTableIdInfo->tid = htonl(pTableMeta->sid);
@@ -618,8 +632,9 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, index);
     
     // set the vgroup info
-    tscSetDnodeIpList(pSql, &pTableIdList->vgInfo);
-    pQueryMsg->head.vgId = htonl(pTableIdList->vgInfo.vgId);
+    int32_t vgId = 0;
+    tscDumpIpSetFromVgroupInfo(&pTableIdList->vgInfo, &pSql->ipList, &vgId);
+    pQueryMsg->head.vgId = htonl(vgId);
     
     int32_t numOfTables = taosArrayGetSize(pTableIdList->itemList);
     pQueryMsg->numOfTables = htonl(numOfTables);  // set the number of tables
@@ -1351,7 +1366,7 @@ int tscBuildUpdateTagMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
   SQueryInfo *    pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
 
-  tscSetDnodeIpList(pSql, &pTableMetaInfo->pTableMeta->vgroupInfo);
+  tscDumpIpSetFromVgroupInfo(&pTableMetaInfo->pTableMeta->vgroupInfo, &pSql->ipList, NULL);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -1875,11 +1890,11 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
 
     memcpy(pInfo->vgroupList, pVgroupInfo, size);
     for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) {
+      //just init, no need to lock
       SCMVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
-
+      pVgroups->version = 0;
       pVgroups->vgId = htonl(pVgroups->vgId);
       assert(pVgroups->numOfIps >= 1);
-
       for (int32_t k = 0; k < pVgroups->numOfIps; ++k) {
         pVgroups->ipAddr[k].port = htons(pVgroups->ipAddr[k].port);
       }
diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h
index b7afaf1e06..c12aed2bcc 100644
--- a/src/inc/taosmsg.h
+++ b/src/inc/taosmsg.h
@@ -647,6 +647,7 @@ typedef struct SCMSTableVgroupMsg {
 } SCMSTableVgroupMsg, SCMSTableVgroupRspMsg;
 
 typedef struct {
+  int32_t   version;
   int32_t   vgId;
   int8_t    numOfIps;
   SIpAddr   ipAddr[TSDB_MAX_REPLICA];

From dd45e955dc5ed8c2a9862abbcda319ced4531f68 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 06:53:56 +0000
Subject: [PATCH 03/25] fix bug

---
 src/client/src/tscServer.c | 20 ++++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 4a11288a85..045aab61d6 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -44,9 +44,18 @@ void tscSaveSubscriptionProgress(void* sub);
 
 static int32_t minMsgSize() { return tsRpcHeadSize + 100; }
 
+void tscIpSetCopy(SRpcIpSet *dst, SRpcIpSet *src) {
+  dst->numOfIps = src->numOfIps;
+  dst->inUse = src->inUse;
+  for (int32_t i = 0; i < src->numOfIps; ++i) {
+     dst->port[i] = htons(dst->port[i]);
+     strncpy(dst->fqdn[i], src->fqdn[i], TSDB_FQDN_LEN);
+  }
+}
 static void tscDumpMgmtIpSet(SRpcIpSet *ipSet) {
   taosCorBeginRead(&tscMgmtIpSet.version);
-  *ipSet = tscMgmtIpSet.ipSet;
+  SRpcIpSet*  src = &tscMgmtIpSet.ipSet;
+  tscIpSetCopy(ipSet, src); 
   taosCorEndRead(&tscMgmtIpSet.version);
 }  
 
@@ -71,12 +80,7 @@ void tscSetMgmtIpList(SRpcIpSet *pIpSet) {
    
   taosCorBeginWrite(&tscMgmtIpSet.version);
   SRpcIpSet *mgmtIpSet = &tscMgmtIpSet.ipSet;
-  mgmtIpSet->numOfIps = pIpSet->numOfIps;
-  mgmtIpSet->inUse = pIpSet->inUse;
-  for (int32_t i = 0; i < mgmtIpSet->numOfIps; ++i) {
-     mgmtIpSet->port[i] = htons(pIpSet->port[i]);
-     strncpy(mgmtIpSet->fqdn[i], pIpSet->fqdn[i], TSDB_FQDN_LEN);
-  }
+  tscIpSetCopy(mgmtIpSet, pIpSet);
   taosCorEndWrite(&tscMgmtIpSet.version);
 }
 static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet, int32_t *vgId) {
@@ -227,7 +231,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
   if (pSql->cmd.command >= TSDB_SQL_MGMT) {
     SRpcIpSet dump;
     tscDumpMgmtIpSet(&dump);
-    pSql->ipList = dump;
+    tscIpSetCopy(&pSql->ipList, &dump);
   }
 
   memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);

From 6163e2401f5d18822c206475543a14019991c500 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 07:42:04 +0000
Subject: [PATCH 04/25] refor code

---
 src/client/src/tscServer.c | 41 ++++++++++++++++++--------------------
 1 file changed, 19 insertions(+), 22 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 045aab61d6..c82773a125 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -70,15 +70,10 @@ bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
    }
    return true;
 }
-void tscSetMgmtIpList(SRpcIpSet *pIpSet) {
+void tscUpdateMgmtIpList(SRpcIpSet *pIpSet) {
   // no need to update if equal
-  SRpcIpSet dump;
-  tscDumpMgmtIpSet(&dump);
-  if (tscIpSetIsEqual(&dump, pIpSet)) {
-    return; 
-  }
-   
   taosCorBeginWrite(&tscMgmtIpSet.version);
+  // or copy directly,  tscMgmtIpSet.ipSet = *pIpSet
   SRpcIpSet *mgmtIpSet = &tscMgmtIpSet.ipSet;
   tscIpSetCopy(mgmtIpSet, pIpSet);
   taosCorEndWrite(&tscMgmtIpSet.version);
@@ -97,7 +92,12 @@ static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pI
   }
   taosCorEndRead(&pVgroupInfo->version);
 }
-static void tscSetVgroupInfoWithIpSet(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet) {
+
+static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
+  SSqlCmd *pCmd = &pObj->cmd;
+  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
+  SCMVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->vgroupInfo;
+
   taosCorBeginWrite(&pVgroupInfo->version);
   //TODO(dengyihao), dont care vgid  
   pVgroupInfo->numOfIps = pIpSet->numOfIps;
@@ -107,14 +107,6 @@ static void tscSetVgroupInfoWithIpSet(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIp
   }
   taosCorEndWrite(&pVgroupInfo->version);
 }
-static void tscSetVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
-  if (tscIpSetIsEqual(&pObj->ipList, pIpSet)) {
-    return;
-  }
-  SSqlCmd *pCmd = &pObj->cmd;
-  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
-  tscSetVgroupInfoWithIpSet(&pTableMetaInfo->pTableMeta->vgroupInfo, pIpSet);  
-}
 void tscPrintMgmtIp() {
   SRpcIpSet dump;
   tscDumpMgmtIpSet(&dump);
@@ -157,7 +149,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
     SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
     SRpcIpSet *      pIpList = &pRsp->ipList;
     if (pIpList->numOfIps > 0) 
-      tscSetMgmtIpList(pIpList);
+      tscUpdateMgmtIpList(pIpList);
 
     pSql->pTscObj->connId = htonl(pRsp->connId);
 
@@ -231,6 +223,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
   if (pSql->cmd.command >= TSDB_SQL_MGMT) {
     SRpcIpSet dump;
     tscDumpMgmtIpSet(&dump);
+    // no need to update pSql->ipList
     tscIpSetCopy(&pSql->ipList, &dump);
   }
 
@@ -282,10 +275,14 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
     return;
   }
 
-  if (pCmd->command < TSDB_SQL_MGMT) {
-    if (pIpSet) tscSetVgroupInfo(pSql, pIpSet); 
-  } else {
-    if (pIpSet) tscSetMgmtIpList(pIpSet);  
+  if (pIpSet) {
+    if (!tscIpSetIsEqual(&pSql->ipList, pIpSet)) {
+      if (pCmd->command < TSDB_SQL_MGMT)  { 
+        tscUpdateVgroupInfo(pSql, pIpSet); 
+      } else {
+        tscUpdateMgmtIpList(pIpSet);
+      }
+    }
   }
 
   if (rpcMsg->pCont == NULL) {
@@ -1994,7 +1991,7 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
   tstrncpy(pObj->db, temp, sizeof(pObj->db));
   
   if (pConnect->ipList.numOfIps > 0) 
-    tscSetMgmtIpList(&pConnect->ipList);
+    tscUpdateMgmtIpList(&pConnect->ipList);
 
   strcpy(pObj->sversion, pConnect->serverVersion);
   pObj->writeAuth = pConnect->writeAuth;

From 5f55393464f57939a17043d3a4a23d3f81ad8353 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 21:54:16 +0000
Subject: [PATCH 05/25] fixbug

---
 src/client/src/tscServer.c | 26 ++++++++++++++++----------
 1 file changed, 16 insertions(+), 10 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index c82773a125..ced6914df2 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -48,7 +48,7 @@ void tscIpSetCopy(SRpcIpSet *dst, SRpcIpSet *src) {
   dst->numOfIps = src->numOfIps;
   dst->inUse = src->inUse;
   for (int32_t i = 0; i < src->numOfIps; ++i) {
-     dst->port[i] = htons(dst->port[i]);
+     dst->port[i] = src->port[i];
      strncpy(dst->fqdn[i], src->fqdn[i], TSDB_FQDN_LEN);
   }
 }
@@ -58,7 +58,11 @@ static void tscDumpMgmtIpSet(SRpcIpSet *ipSet) {
   tscIpSetCopy(ipSet, src); 
   taosCorEndRead(&tscMgmtIpSet.version);
 }  
-
+static void tscIpSetHtons(SRpcIpSet *s) {
+   for (int32_t i = 0; i < s->numOfIps; i++) {
+      s->port[i] = htons(s->port[i]);    
+   }
+} 
 bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
    if (s1->numOfIps != s2->numOfIps /*|| s1->inUse != s1->inUse*/) {
      return false;
@@ -149,6 +153,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
     SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
     SRpcIpSet *      pIpList = &pRsp->ipList;
     if (pIpList->numOfIps > 0) 
+      tscIpSetHtons(pIpList);
       tscUpdateMgmtIpList(pIpList);
 
     pSql->pTscObj->connId = htonl(pRsp->connId);
@@ -275,13 +280,15 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
     return;
   }
 
-  if (pIpSet) {
-    if (!tscIpSetIsEqual(&pSql->ipList, pIpSet)) {
-      if (pCmd->command < TSDB_SQL_MGMT)  { 
+  if (pIpSet) { 
+    //SRpcIpSet dump; 
+    tscIpSetHtons(pIpSet); 
+    if (tscIpSetIsEqual(&pSql->ipList, pIpSet)) {
+      if(pCmd->command < TSDB_SQL_MGMT)  { 
         tscUpdateVgroupInfo(pSql, pIpSet); 
       } else {
         tscUpdateMgmtIpList(pIpSet);
-      }
+    }
     }
   }
 
@@ -464,10 +471,8 @@ int tscProcessSql(SSqlObj *pSql) {
       return pSql->res.code;
     }
   } else if (pCmd->command < TSDB_SQL_LOCAL) {
-    SRpcIpSet dump;
-    tscDumpMgmtIpSet(&dump);
-    pSql->ipList = dump;
-  } else {  // local handler
+    tscDumpMgmtIpSet(&pSql->ipList);
+  } else {  
     return (*tscProcessMsgRsp[pCmd->command])(pSql);
   }
   
@@ -1991,6 +1996,7 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
   tstrncpy(pObj->db, temp, sizeof(pObj->db));
   
   if (pConnect->ipList.numOfIps > 0) 
+    tscIpSetHtons(&pConnect->ipList);
     tscUpdateMgmtIpList(&pConnect->ipList);
 
   strcpy(pObj->sversion, pConnect->serverVersion);

From d2823d60e2bc30dffa04970bf970de53761f076a Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 22:17:23 +0000
Subject: [PATCH 06/25] modify code style

---
 src/client/src/tscServer.c | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index ced6914df2..b162387685 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -226,10 +226,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
 
   // set the mgmt ip list
   if (pSql->cmd.command >= TSDB_SQL_MGMT) {
-    SRpcIpSet dump;
-    tscDumpMgmtIpSet(&dump);
-    // no need to update pSql->ipList
-    tscIpSetCopy(&pSql->ipList, &dump);
+    tscDumpMgmtIpSet(&pSql->ipList);
   }
 
   memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
@@ -471,7 +468,7 @@ int tscProcessSql(SSqlObj *pSql) {
       return pSql->res.code;
     }
   } else if (pCmd->command < TSDB_SQL_LOCAL) {
-    tscDumpMgmtIpSet(&pSql->ipList);
+    //tscDumpMgmtIpSet(&pSql->ipList);
   } else {  
     return (*tscProcessMsgRsp[pCmd->command])(pSql);
   }

From 65b32c3855ecfd7caf2aef977ada081e5e3b1b66 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Fri, 17 Jul 2020 22:35:57 +0000
Subject: [PATCH 07/25] modify code style

---
 src/client/src/tscServer.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index b162387685..3b3884eb07 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -100,6 +100,7 @@ static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pI
 static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
   SSqlCmd *pCmd = &pObj->cmd;
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
+  if (pTableMetaInfo == NULL || pTableMetaInfo->pTableMeta == NULL) { return;}
   SCMVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->vgroupInfo;
 
   taosCorBeginWrite(&pVgroupInfo->version);

From 0ac592fbfd44686a875629a93760d6e48e24b8bf Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 01:04:02 +0000
Subject: [PATCH 08/25] add inuse to vgroupinfo

---
 src/client/src/tscServer.c | 9 ++++++---
 src/inc/taosmsg.h          | 1 +
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 3b3884eb07..d0c9d3636b 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -64,7 +64,7 @@ static void tscIpSetHtons(SRpcIpSet *s) {
    }
 } 
 bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
-   if (s1->numOfIps != s2->numOfIps /*|| s1->inUse != s1->inUse*/) {
+   if (s1->numOfIps != s2->numOfIps || s1->inUse != s2->inUse) {
      return false;
    } 
    for (int32_t i = 0; i < s1->numOfIps; i++) {
@@ -88,7 +88,8 @@ static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pI
   if (vgId) { 
     *vgId = pVgroupInfo->vgId; 
   }
-  pIpSet->inUse = 0; 
+  int8_t inUse = pVgroupInfo->inUse;
+  pIpSet->inUse = (inUse >= 0 && inUse < TSDB_MAX_REPLICA) ? inUse: 0; 
   pIpSet->numOfIps = pVgroupInfo->numOfIps;  
   for (int32_t i = 0; i < pVgroupInfo->numOfIps; ++i) {
     strncpy(pIpSet->fqdn[i], pVgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
@@ -104,7 +105,8 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
   SCMVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->vgroupInfo;
 
   taosCorBeginWrite(&pVgroupInfo->version);
-  //TODO(dengyihao), dont care vgid  
+  //TODO(dengyihao), dont care vgid 
+  pVgroupInfo->inUse = pIpSet->inUse;
   pVgroupInfo->numOfIps = pIpSet->numOfIps;
   for (int32_t i = 0; pVgroupInfo->numOfIps; i++) {
     strncpy(pVgroupInfo->ipAddr[i].fqdn, pIpSet->fqdn[i], TSDB_FQDN_LEN);
@@ -1897,6 +1899,7 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
       //just init, no need to lock
       SCMVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
       pVgroups->version = 0;
+      pVgroups->inUse = 0;
       pVgroups->vgId = htonl(pVgroups->vgId);
       assert(pVgroups->numOfIps >= 1);
       for (int32_t k = 0; k < pVgroups->numOfIps; ++k) {
diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h
index c12aed2bcc..fec3d77d8e 100644
--- a/src/inc/taosmsg.h
+++ b/src/inc/taosmsg.h
@@ -649,6 +649,7 @@ typedef struct SCMSTableVgroupMsg {
 typedef struct {
   int32_t   version;
   int32_t   vgId;
+  int8_t    inUse;
   int8_t    numOfIps;
   SIpAddr   ipAddr[TSDB_MAX_REPLICA];
 } SCMVgroupInfo;

From f569bb4babb1341c4128e24cfe2c20005cca011e Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 17 Jul 2020 14:27:02 +0800
Subject: [PATCH 09/25] td-909: fix subscription bugs

---
 src/client/src/tscSub.c   | 164 ++++++++++++++++++++++----------------
 src/query/src/qExecutor.c |   2 +-
 2 files changed, 97 insertions(+), 69 deletions(-)

diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c
index 2c5035c2ef..73614fa0a5 100644
--- a/src/client/src/tscSub.c
+++ b/src/client/src/tscSub.c
@@ -34,6 +34,7 @@ typedef struct SSubscriptionProgress {
 typedef struct SSub {
   void *                  signature;
   char                    topic[32];
+  sem_t                   sem;
   int64_t                 lastSyncTime;
   int64_t                 lastConsumeTime;
   TAOS *                  taos;
@@ -83,84 +84,107 @@ void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts) {
 
 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);
+  SSub *pSub = ((SSub *)param);
+  pSub->pSql->res.code = code;
+  sem_post(&pSub->sem);
 }
 
 
 static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* sql) {
-  SSub* pSub = NULL;
+  int code = TSDB_CODE_SUCCESS, line = __LINE__;
 
-  TRY( 8 ) {
-    SSqlObj* pSql = calloc_throw(1, sizeof(SSqlObj));
-    CLEANUP_PUSH_FREE(true, pSql);
-    SSqlCmd *pCmd = &pSql->cmd;
-    SSqlRes *pRes = &pSql->res;
+  SSub* pSub = calloc(1, sizeof(SSub));
+  if (pSub == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  pSub->signature = pSub;
+  if (tsem_init(&pSub->sem, 0, 0) == -1) {
+    line = __LINE__;
+    code = TAOS_SYSTEM_ERROR(errno);
+    goto fail;
+  }
+  tstrncpy(pSub->topic, topic, sizeof(pSub->topic));
+  pSub->progress = taosArrayInit(32, sizeof(SSubscriptionProgress));
+  if (pSub->progress == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
 
-    if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
-      THROW(TAOS_SYSTEM_ERROR(errno));
-    }
-    CLEANUP_PUSH_INT_PTR(true, tsem_destroy, &pSql->rspSem);
+  SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
+  if (pSql == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  pSql->signature = pSql;
+  pSql->pTscObj = pObj;
+  pSql->pSubscription = pSub;
+  pSub->pSql = pSql;
 
-    pSql->signature = pSql;
-    pSql->param = pSql;
-    pSql->pTscObj = pObj;
-    pSql->maxRetry = TSDB_MAX_REPLICA;
-    pSql->fp = asyncCallback;
+  SSqlCmd* pCmd = &pSql->cmd;
+  SSqlRes* pRes = &pSql->res;
+  if (tsem_init(&pSql->rspSem, 0, 0) == -1) {
+    line = __LINE__;
+    code = TAOS_SYSTEM_ERROR(errno);
+    goto fail;
+  }
 
-    int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
-    if (code != TSDB_CODE_SUCCESS) {
-      THROW(code);
-    }
-    CLEANUP_PUSH_FREE(true, pCmd->payload);
+  pSql->param = pSub;
+  pSql->maxRetry = TSDB_MAX_REPLICA;
+  pSql->fp = asyncCallback;
+  pSql->fetchFp = asyncCallback;
+  pSql->sqlstr = strdup(sql);
+  if (pSql->sqlstr == NULL) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto fail;
+  }
+  strtolower(pSql->sqlstr, pSql->sqlstr);
+  pRes->qhandle = 0;
+  pRes->numOfRows = 1;
 
-    pRes->qhandle = 0;
-    pRes->numOfRows = 1;
+  code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
+  if (code != TSDB_CODE_SUCCESS) {
+    line = __LINE__;
+    goto fail;
+  }
 
-    pSql->sqlstr = strdup_throw(sql);
-    CLEANUP_PUSH_FREE(true, pSql->sqlstr);
-    strtolower(pSql->sqlstr, pSql->sqlstr);
+  code = tsParseSql(pSql, false);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    sem_wait(&pSub->sem);
+    code = pSql->res.code;
+  }
+  if (code != TSDB_CODE_SUCCESS) {
+    line = __LINE__;
+    goto fail;
+  }
 
-    code = tsParseSql(pSql, false);
-    if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
-      // wait for the callback function to post the semaphore
-      sem_wait(&pSql->rspSem);
-      code = pSql->res.code;
-    }
-    if (code != TSDB_CODE_SUCCESS) {
-      tscError("failed to parse sql statement: %s, error: %s", pSub->topic, tstrerror(code));
-      THROW( code );
-    }
-
-    if (pSql->cmd.command != TSDB_SQL_SELECT) {
-      tscError("only 'select' statement is allowed in subscription: %s", pSub->topic);
-      THROW( -1 );  // TODO
-    }
-
-    pSub = calloc_throw(1, sizeof(SSub));
-    CLEANUP_PUSH_FREE(true, pSub);
-    pSql->pSubscription = pSub;
-    pSub->pSql = pSql;
-    pSub->signature = pSub;
-    strncpy(pSub->topic, topic, sizeof(pSub->topic));
-    pSub->topic[sizeof(pSub->topic) - 1] = 0;
-    pSub->progress = taosArrayInit(32, sizeof(SSubscriptionProgress));
-    if (pSub->progress == NULL) {
-      THROW(TSDB_CODE_TSC_OUT_OF_MEMORY);
-    }
-
-    CLEANUP_EXECUTE();
-
-  } CATCH( code ) {
-    tscError("failed to create subscription object: %s", tstrerror(code));
-    CLEANUP_EXECUTE();
-    pSub = NULL;
-
-  } END_TRY
+  if (pSql->cmd.command != TSDB_SQL_SELECT) {
+    line = __LINE__;
+    code = TSDB_CODE_TSC_INVALID_SQL;
+    goto fail;
+  }
 
   return pSub;
+
+fail:
+  tscError("tscCreateSubscription failed at line %d, reason: %s", line, tstrerror(code));
+  if (pSql != NULL) {
+    tscFreeSqlObj(pSql);
+    pSql = NULL;
+  }
+  if (pSub != NULL) {
+    taosArrayDestroy(pSub->progress);
+    tsem_destroy(&pSub->sem);
+    free(pSub);
+    pSub = NULL;
+  }
+
+  terrno = code;
+  return NULL;
 }
 
 
@@ -405,9 +429,10 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
     tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0)->vgroupIndex = 0;
 
     pSql->fp = asyncCallback;
-    pSql->param = pSql;
+    pSql->fetchFp = asyncCallback;
+    pSql->param = pSub;
     tscDoQuery(pSql);
-    sem_wait(&pSql->rspSem);
+    sem_wait(&pSub->sem);
 
     if (pRes->code != TSDB_CODE_SUCCESS) {
       continue;
@@ -437,7 +462,9 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
   }
 
   if (keepProgress) {
-    tscSaveSubscriptionProgress(pSub);
+    if (pSub->progress != NULL) {
+      tscSaveSubscriptionProgress(pSub);
+    }
   } else {
     char path[256];
     sprintf(path, "%s/subscribe/%s", tsDataDir, pSub->topic);
@@ -448,6 +475,7 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
 
   tscFreeSqlObj(pSub->pSql);
   taosArrayDestroy(pSub->progress);
+  tsem_destroy(&pSub->sem);
   memset(pSub, 0, sizeof(*pSub));
   free(pSub);
 }
diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c
index cf13ace11f..3b10967e33 100644
--- a/src/query/src/qExecutor.c
+++ b/src/query/src/qExecutor.c
@@ -4918,7 +4918,7 @@ static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
         pQuery->current->lastKey, pQuery->window.ekey);
   } else if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
     STableIdInfo tidInfo;
-    STableId* id = TSDB_TABLEID(pQuery->current);
+    STableId* id = TSDB_TABLEID(pQuery->current->pTable);
 
     tidInfo.uid = id->uid;
     tidInfo.tid = id->tid;

From f9cb99e6f79e0cab9a40ad4840c49f348f2392b8 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 17 Jul 2020 15:47:20 +0800
Subject: [PATCH 10/25] td-909: make taos_validate_sql work

---
 src/client/src/tscSql.c | 25 ++++++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c
index d5160cd3c6..49e4e027a0 100644
--- a/src/client/src/tscSql.c
+++ b/src/client/src/tscSql.c
@@ -724,6 +724,13 @@ int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields)
   return len;
 }
 
+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);
+}
+
 int taos_validate_sql(TAOS *taos, const char *sql) {
   STscObj *pObj = (STscObj *)taos;
   if (pObj == NULL || pObj->signature != pObj) {
@@ -732,7 +739,8 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
   }
 
   SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
-  
+  pSql->pTscObj = taos;
+  pSql->signature = pSql;
   SSqlRes *pRes = &pSql->res;
   SSqlCmd *pCmd = &pSql->cmd;
   
@@ -766,10 +774,17 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
     pCmd->pTableList = NULL;
   }
 
-  pRes->code = (uint8_t)tsParseSql(pSql, false);
-  int code = pRes->code;
-
-  tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj);
+  pSql->fp = asyncCallback;
+  pSql->fetchFp = asyncCallback;
+  pSql->param = pSql;
+  int code = tsParseSql(pSql, true);
+  if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    sem_wait(&pSql->rspSem);
+    code = pSql->res.code;
+  }
+  if (code != TSDB_CODE_SUCCESS) {
+    tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, code, taos_errstr(taos), pObj);
+  }
   taos_free_result(pSql);
 
   return code;

From efe558731b14053a04cbfb84ea1e1000fce0881b Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 17 Jul 2020 15:56:49 +0800
Subject: [PATCH 11/25] td-909: add verification for some API

---
 tests/examples/c/apitest.c | 394 +++++++++++++++++++++++++++++++++++++
 1 file changed, 394 insertions(+)
 create mode 100644 tests/examples/c/apitest.c

diff --git a/tests/examples/c/apitest.c b/tests/examples/c/apitest.c
new file mode 100644
index 0000000000..0be1f01bbe
--- /dev/null
+++ b/tests/examples/c/apitest.c
@@ -0,0 +1,394 @@
+// sample code to verify all TDengine API
+// to compile: gcc -o apitest apitest.c -ltaos
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <taos.h>
+#include <unistd.h>
+
+
+static void prepare_data(TAOS* taos) {
+  taos_query(taos, "drop database if exists test;");
+  usleep(100000);
+  taos_query(taos, "create database test;");
+  usleep(100000);
+  taos_select_db(taos, "test");
+
+  taos_query(taos, "create table meters(ts timestamp, a int) tags(area int);");
+
+  taos_query(taos, "create table t0 using meters tags(0);");
+  taos_query(taos, "create table t1 using meters tags(1);");
+  taos_query(taos, "create table t2 using meters tags(2);");
+  taos_query(taos, "create table t3 using meters tags(3);");
+  taos_query(taos, "create table t4 using meters tags(4);");
+  taos_query(taos, "create table t5 using meters tags(5);");
+  taos_query(taos, "create table t6 using meters tags(6);");
+  taos_query(taos, "create table t7 using meters tags(7);");
+  taos_query(taos, "create table t8 using meters tags(8);");
+  taos_query(taos, "create table t9 using meters tags(9);");
+
+  TAOS_RES* res = taos_query(taos, "insert into t0 values('2020-01-01 00:00:00.000', 0)"
+    " ('2020-01-01 00:01:00.000', 0)"
+    " ('2020-01-01 00:02:00.000', 0)"
+    " t1 values('2020-01-01 00:00:00.000', 0)"
+    " ('2020-01-01 00:01:00.000', 0)"
+    " ('2020-01-01 00:02:00.000', 0)"
+    " ('2020-01-01 00:03:00.000', 0)"
+    " t2 values('2020-01-01 00:00:00.000', 0)"
+    " ('2020-01-01 00:01:00.000', 0)"
+    " ('2020-01-01 00:01:01.000', 0)"
+    " ('2020-01-01 00:01:02.000', 0)"
+    " t3 values('2020-01-01 00:01:02.000', 0)"
+    " t4 values('2020-01-01 00:01:02.000', 0)"
+    " t5 values('2020-01-01 00:01:02.000', 0)"
+    " t6 values('2020-01-01 00:01:02.000', 0)"
+    " t7 values('2020-01-01 00:01:02.000', 0)"
+    " t8 values('2020-01-01 00:01:02.000', 0)"
+    " t9 values('2020-01-01 00:01:02.000', 0)");
+  int affected = taos_affected_rows(res);
+  if (affected != 18) {
+    printf("\033[31m%d rows affected by last insert statement, but it should be 18\033[0m\n", affected);
+  }
+  // super tables subscription
+  usleep(1000000);
+}
+
+
+static int print_result(TAOS_RES* res, int blockFetch) {
+  TAOS_ROW    row = NULL;
+  int         num_fields = taos_num_fields(res);
+  TAOS_FIELD* fields = taos_fetch_fields(res);
+  int         nRows = 0;
+
+  if (blockFetch) {
+    int rows = 0;
+    while ((rows = taos_fetch_block(res, &row))) {
+      for (int i = 0; i < rows; i++) {
+        char temp[256];
+        taos_print_row(temp, row + i, fields, num_fields);
+        puts(temp);
+      }
+      nRows += rows;
+    }
+  } else {
+    while ((row = taos_fetch_row(res))) {
+      char temp[256];
+      taos_print_row(temp, row, fields, num_fields);
+      puts(temp);
+      nRows++;
+    }
+  }
+
+  printf("%d rows consumed.\n", nRows);
+  return nRows;
+}
+
+
+static void check_row_count(int line, TAOS_RES* res, int expected) {
+  int actual = print_result(res, expected % 2);
+  if (actual != expected) {
+    printf("\033[31mline %d: row count mismatch, expected: %d, actual: %d\033[0m\n", line, expected, actual);
+  } else {
+    printf("line %d: %d rows consumed as expected\n", line, actual);
+  }
+}
+
+
+static void verify_query(TAOS* taos) {
+  prepare_data(taos);
+
+  int code = taos_validate_sql(taos, "select * from nonexisttable");
+  if (code == 0) {
+    printf("\033[31mimpossible, the table does not exists\033[0m\n");
+  }
+
+  code = taos_validate_sql(taos, "select * from meters");
+  if (code != 0) {
+    printf("\033[31mimpossible, the table does exists: 0x%08x\033[0m\n", code);
+  }
+
+  TAOS_RES* res = taos_query(taos, "select * from meters");
+  check_row_count(__LINE__, res, 18);
+  printf("result precision is: %d\n",  taos_result_precision(res));
+  int c = taos_field_count(res);
+  printf("field count is: %d\n",  c);
+  int* lengths = taos_fetch_lengths(res);
+  for (int i = 0; i < c; i++) {
+    printf("length of column %d is %d\n", i, lengths[i]);
+  }
+  taos_free_result(res);
+
+  res = taos_query(taos, "select * from t0");
+  check_row_count(__LINE__, res, 3);
+  taos_free_result(res);
+
+  res = taos_query(taos, "select * from nonexisttable");
+  code = taos_errno(res);
+  printf("code=%d, error msg=%s\n", code, taos_errstr(res));
+  taos_free_result(res);
+}
+
+
+static void verify_subscribe(TAOS* taos) {
+  prepare_data(taos);
+
+  TAOS_SUB* tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0);
+  TAOS_RES* res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 18);
+
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 0);
+
+  taos_query(taos, "insert into t0 values('2020-01-01 00:02:00.001', 0);");
+  taos_query(taos, "insert into t8 values('2020-01-01 00:01:03.000', 0);");
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 2);
+
+  taos_query(taos, "insert into t2 values('2020-01-01 00:01:02.001', 0);");
+  taos_query(taos, "insert into t1 values('2020-01-01 00:03:00.001', 0);");
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 2);
+
+  taos_query(taos, "insert into t1 values('2020-01-01 00:03:00.002', 0);");
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 1);
+
+  // keep progress information and restart subscription
+  taos_unsubscribe(tsub, 1);
+  taos_query(taos, "insert into t0 values('2020-01-01 00:04:00.000', 0);");
+  tsub = taos_subscribe(taos, 1, "test", "select * from meters;", NULL, NULL, 0);
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 24);
+
+  // keep progress information and continue previous subscription
+  taos_unsubscribe(tsub, 1);
+  tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0);
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 0);
+
+  // don't keep progress information and continue previous subscription
+  taos_unsubscribe(tsub, 0);
+  tsub = taos_subscribe(taos, 0, "test", "select * from meters;", NULL, NULL, 0);
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 24);
+
+  // single meter subscription
+
+  taos_unsubscribe(tsub, 0);
+  tsub = taos_subscribe(taos, 0, "test", "select * from t0;", NULL, NULL, 0);
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 5);
+
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 0);
+
+  taos_query(taos, "insert into t0 values('2020-01-01 00:04:00.001', 0);");
+  res = taos_consume(tsub);
+  check_row_count(__LINE__, res, 1);
+
+  taos_unsubscribe(tsub, 0);
+}
+
+
+void verify_prepare(TAOS* taos) {
+  TAOS_RES* result = taos_query(taos, "drop database if exists test;");
+  usleep(100000);
+  taos_query(taos, "create database test;");
+
+  int code = taos_errno(result);
+  if (code != 0) {
+    printf("failed to create database, reason:%s\n", taos_errstr(result));
+    taos_free_result(result);
+    exit(1);
+  }
+  taos_free_result(result);
+
+  usleep(100000);
+  taos_select_db(taos, "test");
+
+  // create table
+  const char* sql = "create table m1 (ts timestamp, b bool, v1 tinyint, v2 smallint, v4 int, v8 bigint, f4 float, f8 double, bin binary(40), blob nchar(10))";
+  result = taos_query(taos, sql);
+  code = taos_errno(result);
+  if (code != 0) {
+    printf("failed to create table, reason:%s\n", taos_errstr(result));
+    taos_free_result(result);
+    exit(1);
+  }
+  taos_free_result(result);
+
+  // insert 10 records
+  struct {
+      int64_t ts;
+      int8_t b;
+      int8_t v1;
+      int16_t v2;
+      int32_t v4;
+      int64_t v8;
+      float f4;
+      double f8;
+      char bin[40];
+      char blob[80];
+  } v = {0};
+
+  TAOS_STMT* stmt = taos_stmt_init(taos);
+  TAOS_BIND params[10];
+  params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
+  params[0].buffer_length = sizeof(v.ts);
+  params[0].buffer = &v.ts;
+  params[0].length = &params[0].buffer_length;
+  params[0].is_null = NULL;
+
+  params[1].buffer_type = TSDB_DATA_TYPE_BOOL;
+  params[1].buffer_length = sizeof(v.b);
+  params[1].buffer = &v.b;
+  params[1].length = &params[1].buffer_length;
+  params[1].is_null = NULL;
+
+  params[2].buffer_type = TSDB_DATA_TYPE_TINYINT;
+  params[2].buffer_length = sizeof(v.v1);
+  params[2].buffer = &v.v1;
+  params[2].length = &params[2].buffer_length;
+  params[2].is_null = NULL;
+
+  params[3].buffer_type = TSDB_DATA_TYPE_SMALLINT;
+  params[3].buffer_length = sizeof(v.v2);
+  params[3].buffer = &v.v2;
+  params[3].length = &params[3].buffer_length;
+  params[3].is_null = NULL;
+
+  params[4].buffer_type = TSDB_DATA_TYPE_INT;
+  params[4].buffer_length = sizeof(v.v4);
+  params[4].buffer = &v.v4;
+  params[4].length = &params[4].buffer_length;
+  params[4].is_null = NULL;
+
+  params[5].buffer_type = TSDB_DATA_TYPE_BIGINT;
+  params[5].buffer_length = sizeof(v.v8);
+  params[5].buffer = &v.v8;
+  params[5].length = &params[5].buffer_length;
+  params[5].is_null = NULL;
+
+  params[6].buffer_type = TSDB_DATA_TYPE_FLOAT;
+  params[6].buffer_length = sizeof(v.f4);
+  params[6].buffer = &v.f4;
+  params[6].length = &params[6].buffer_length;
+  params[6].is_null = NULL;
+
+  params[7].buffer_type = TSDB_DATA_TYPE_DOUBLE;
+  params[7].buffer_length = sizeof(v.f8);
+  params[7].buffer = &v.f8;
+  params[7].length = &params[7].buffer_length;
+  params[7].is_null = NULL;
+
+  params[8].buffer_type = TSDB_DATA_TYPE_BINARY;
+  params[8].buffer_length = sizeof(v.bin);
+  params[8].buffer = v.bin;
+  params[8].length = &params[8].buffer_length;
+  params[8].is_null = NULL;
+
+  strcpy(v.blob, "一二三四五六七八九十");
+  params[9].buffer_type = TSDB_DATA_TYPE_NCHAR;
+  params[9].buffer_length = strlen(v.blob);
+  params[9].buffer = v.blob;
+  params[9].length = &params[9].buffer_length;
+  params[9].is_null = NULL;
+
+  int is_null = 1;
+
+  sql = "insert into m1 values(?,?,?,?,?,?,?,?,?,?)";
+  code = taos_stmt_prepare(stmt, sql, 0);
+  if (code != 0){
+    printf("failed to execute taos_stmt_prepare. code:0x%x\n", code);
+  }
+  v.ts = 1591060628000;
+  for (int i = 0; i < 10; ++i) {
+    v.ts += 1;
+    for (int j = 1; j < 10; ++j) {
+      params[j].is_null = ((i == j) ? &is_null : 0);
+    }
+    v.b = (int8_t)i % 2;
+    v.v1 = (int8_t)i;
+    v.v2 = (int16_t)(i * 2);
+    v.v4 = (int32_t)(i * 4);
+    v.v8 = (int64_t)(i * 8);
+    v.f4 = (float)(i * 40);
+    v.f8 = (double)(i * 80);
+    for (int j = 0; j < sizeof(v.bin) - 1; ++j) {
+      v.bin[j] = (char)(i + '0');
+    }
+
+    taos_stmt_bind_param(stmt, params);
+    taos_stmt_add_batch(stmt);
+  }
+  if (taos_stmt_execute(stmt) != 0) {
+    printf("failed to execute insert statement.\n");
+    exit(1);
+  }
+  taos_stmt_close(stmt);
+
+  // query the records
+  stmt = taos_stmt_init(taos);
+  taos_stmt_prepare(stmt, "SELECT * FROM m1 WHERE v1 > ? AND v2 < ?", 0);
+  v.v1 = 5;
+  v.v2 = 15;
+  taos_stmt_bind_param(stmt, params + 2);
+  if (taos_stmt_execute(stmt) != 0) {
+    printf("failed to execute select statement.\n");
+    exit(1);
+  }
+
+  result = taos_stmt_use_result(stmt);
+
+  TAOS_ROW    row;
+  int         rows = 0;
+  int         num_fields = taos_num_fields(result);
+  TAOS_FIELD *fields = taos_fetch_fields(result);
+  char        temp[256];
+
+  // fetch the records row by row
+  while ((row = taos_fetch_row(result))) {
+    rows++;
+    taos_print_row(temp, row, fields, num_fields);
+    printf("%s\n", temp);
+  }
+
+  taos_free_result(result);
+  taos_stmt_close(stmt);
+}
+
+void verify_stream(TAOS* taos) {
+}
+
+int main(int argc, char *argv[]) {
+  const char* host = "127.0.0.1";
+  const char* user = "root";
+  const char* passwd = "taosdata";
+
+  taos_options(TSDB_OPTION_TIMEZONE, "GMT-8");
+  taos_init();
+
+  TAOS* taos = taos_connect(host, user, passwd, "", 0);
+  if (taos == NULL) {
+    printf("failed to connect to db, reason:%s\n", taos_errstr(taos));
+    exit(1);
+  }
+
+  char* info = taos_get_server_info(taos);
+  printf("server info: %s\n", info);
+  info = taos_get_client_info(taos);
+  printf("client info: %s\n", info);
+
+  printf("************  verify query  *************\n");
+  verify_query(taos);
+  printf("*********** verify subscribe ************\n");
+  verify_subscribe(taos);
+  printf("************ verify prepare *************\n");
+  verify_prepare(taos);
+  printf("************ verify stream  *************\n");
+  verify_stream(taos);
+
+  taos_close(taos);
+  taos_cleanup();
+}
\ No newline at end of file

From dbd22da833284b240dda7fe32935c1fb9a5009d0 Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Fri, 17 Jul 2020 18:28:33 +0800
Subject: [PATCH 12/25] td-909: fix stream start time issue

fix crash in taos_load_table_info
---
 src/client/src/tscSql.c    |  2 +
 src/client/src/tscStream.c |  6 ++-
 tests/examples/c/apitest.c | 82 +++++++++++++++++++++++++++++++++++++-
 3 files changed, 88 insertions(+), 2 deletions(-)

diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c
index 49e4e027a0..b1c85dd662 100644
--- a/src/client/src/tscSql.c
+++ b/src/client/src/tscSql.c
@@ -880,6 +880,8 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
   }
 
   SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
+  pSql->pTscObj = taos;
+  pSql->signature = pSql;
   SSqlRes *pRes = &pSql->res;
 
   pRes->numOfTotal = 0;  // the number of getting table meta from server
diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c
index 7c188ec969..b07627c87b 100644
--- a/src/client/src/tscStream.c
+++ b/src/client/src/tscStream.c
@@ -122,7 +122,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
       pQueryInfo->window.ekey = pStream->etime;
     }
   } else {
-    pQueryInfo->window.skey = pStream->stime - pStream->interval;
+    pQueryInfo->window.skey = pStream->stime;
     int64_t etime = taosGetTimestamp(pStream->precision);
     // delay to wait all data in last time window
     if (pStream->precision == TSDB_TIME_PRECISION_MICRO) {
@@ -232,6 +232,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
       (*pStream->fp)(pStream->param, res, row);
     }
 
+    if (!pStream->isProject) {
+      pStream->stime += pStream->slidingTime;
+    }
     // actually only one row is returned. this following is not necessary
     taos_fetch_rows_a(res, tscProcessStreamRetrieveResult, pStream);
   } else {  // numOfRows == 0, all data has been retrieved
@@ -432,6 +435,7 @@ static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, in
   } else {             // timewindow based aggregation stream
     if (stime == 0) {  // no data in meter till now
       stime = ((int64_t)taosGetTimestamp(pStream->precision) / pStream->interval) * pStream->interval;
+      stime -= pStream->interval;
       tscWarn("%p stream:%p, last timestamp:0, reset to:%" PRId64, pSql, pStream, stime);
     } else {
       int64_t newStime = (stime / pStream->interval) * pStream->interval;
diff --git a/tests/examples/c/apitest.c b/tests/examples/c/apitest.c
index 0be1f01bbe..bc628fb58f 100644
--- a/tests/examples/c/apitest.c
+++ b/tests/examples/c/apitest.c
@@ -98,7 +98,12 @@ static void check_row_count(int line, TAOS_RES* res, int expected) {
 static void verify_query(TAOS* taos) {
   prepare_data(taos);
 
-  int code = taos_validate_sql(taos, "select * from nonexisttable");
+  int code = taos_load_table_info(taos, "t0,t1,t2,t3,t4,t5,t6,t7,t8,t9");
+  if (code != 0) {
+    printf("\033[31mfailed to load table info: 0x%08x\033[0m\n", code);
+  }
+
+  code = taos_validate_sql(taos, "select * from nonexisttable");
   if (code == 0) {
     printf("\033[31mimpossible, the table does not exists\033[0m\n");
   }
@@ -127,9 +132,17 @@ static void verify_query(TAOS* taos) {
   code = taos_errno(res);
   printf("code=%d, error msg=%s\n", code, taos_errstr(res));
   taos_free_result(res);
+
+  res = taos_query(taos, "select * from meters");
+  taos_stop_query(res);
 }
 
 
+void subscribe_callback(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code) {
+  int rows = print_result(res, *(int*)param);
+  printf("%d rows consumed in subscribe_callback\n", rows);
+}
+
 static void verify_subscribe(TAOS* taos) {
   prepare_data(taos);
 
@@ -188,6 +201,13 @@ static void verify_subscribe(TAOS* taos) {
   check_row_count(__LINE__, res, 1);
 
   taos_unsubscribe(tsub, 0);
+
+  int blockFetch = 0;
+  tsub = taos_subscribe(taos, 1, "test", "select * from meters;", subscribe_callback, &blockFetch, 1000);
+  usleep(2000000);
+  taos_query(taos, "insert into t0 values('2020-01-01 00:05:00.001', 0);");
+  usleep(2000000);
+  taos_unsubscribe(tsub, 0);
 }
 
 
@@ -358,7 +378,60 @@ void verify_prepare(TAOS* taos) {
   taos_stmt_close(stmt);
 }
 
+void retrieve_callback(void *param, TAOS_RES *tres, int numOfRows)
+{
+  if (numOfRows > 0) {
+    printf("%d rows async retrieved\n", numOfRows);
+    taos_fetch_rows_a(tres, retrieve_callback, param);
+  } else {
+    if (numOfRows < 0) {
+      printf("\033[31masync retrieve failed, code: %d\033[0m\n", numOfRows);
+    } else {
+      printf("async retrieve completed\n");
+    }
+    taos_free_result(tres);
+  }
+}
+
+void select_callback(void *param, TAOS_RES *tres, int code)
+{
+  if (code == 0 && tres) {
+    taos_fetch_rows_a(tres, retrieve_callback, param);
+  } else {
+    printf("\033[31masync select failed, code: %d\033[0m\n", code);
+  }
+}
+
+void verify_async(TAOS* taos) {
+  prepare_data(taos);
+  taos_query_a(taos, "select * from meters", select_callback, NULL);
+  usleep(1000000);
+}
+
+void stream_callback(void *param, TAOS_RES *res, TAOS_ROW row) {
+  int         num_fields = taos_num_fields(res);
+  TAOS_FIELD* fields = taos_fetch_fields(res);
+
+  printf("got one row from stream_callback\n");
+  char temp[256];
+  taos_print_row(temp, row, fields, num_fields);
+  puts(temp);
+}
+
 void verify_stream(TAOS* taos) {
+  prepare_data(taos);
+  TAOS_STREAM* strm = taos_open_stream(
+    taos,
+    "select count(*) from meters interval(1m)",
+    stream_callback,
+    0,
+    NULL,
+    NULL);
+  printf("waiting for stream data\n"); 
+  usleep(100000);
+  taos_query(taos, "insert into t0 values(now, 0)(now+5s,1)(now+10s, 2);");
+  usleep(200000000);
+  taos_close_stream(strm);
 }
 
 int main(int argc, char *argv[]) {
@@ -382,12 +455,19 @@ int main(int argc, char *argv[]) {
 
   printf("************  verify query  *************\n");
   verify_query(taos);
+
+  printf("*********  verify async query  **********\n");
+  verify_async(taos);
+
   printf("*********** verify subscribe ************\n");
   verify_subscribe(taos);
+
   printf("************ verify prepare *************\n");
   verify_prepare(taos);
+
   printf("************ verify stream  *************\n");
   verify_stream(taos);
+  printf("done\n");
 
   taos_close(taos);
   taos_cleanup();

From 700504900fc6419f6e87f26d2449ab2168144017 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 01:32:00 +0000
Subject: [PATCH 13/25] compile error

---
 src/client/src/tscServer.c | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index d0c9d3636b..cafdcfbf25 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -155,9 +155,10 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
   if (code == 0) {
     SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
     SRpcIpSet *      pIpList = &pRsp->ipList;
-    if (pIpList->numOfIps > 0) 
+    if (pIpList->numOfIps > 0) {
       tscIpSetHtons(pIpList);
       tscUpdateMgmtIpList(pIpList);
+    } 
 
     pSql->pTscObj->connId = htonl(pRsp->connId);
 
@@ -1996,9 +1997,10 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
   assert(len <= sizeof(pObj->db));
   tstrncpy(pObj->db, temp, sizeof(pObj->db));
   
-  if (pConnect->ipList.numOfIps > 0) 
+  if (pConnect->ipList.numOfIps > 0) {
     tscIpSetHtons(&pConnect->ipList);
     tscUpdateMgmtIpList(&pConnect->ipList);
+  } 
 
   strcpy(pObj->sversion, pConnect->serverVersion);
   pObj->writeAuth = pConnect->writeAuth;

From 54ab10737a7dfc0b433680a16751858077fb6e8b Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 01:43:22 +0000
Subject: [PATCH 14/25] compile error

---
 src/client/src/tscServer.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index cafdcfbf25..9425b32aab 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -26,6 +26,7 @@
 #include "ttime.h"
 #include "ttimer.h"
 #include "tutil.h"
+#include "tlockfree.h"
 
 #define TSC_MGMT_VNODE 999
 

From 7aa062c9ad5679891f0d949160146c02eaa6b71a Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 03:03:04 +0000
Subject: [PATCH 15/25] modify function

---
 src/util/inc/tlockfree.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/util/inc/tlockfree.h b/src/util/inc/tlockfree.h
index e425d71d27..a81f597832 100644
--- a/src/util/inc/tlockfree.h
+++ b/src/util/inc/tlockfree.h
@@ -75,7 +75,7 @@ void taosRUnLockLatch(SRWLatch *pLatch);
 
 // copy on read
 #define taosCorBeginRead(x) for (uint32_t i_ = 1; 1; ++i_) { \
-    int32_t old_ = atomic_load_32(x); \
+    int32_t old_ = atomic_add_fetch_32((x), 0); \
     if (old_ & 0x00000001) { \
       if (i_ % 1000 == 0) { \
         sched_yield(); \
@@ -84,7 +84,7 @@ void taosRUnLockLatch(SRWLatch *pLatch);
     }
 
 #define taosCorEndRead(x) \
-    if (atomic_load_32(x) == old_) { \
+    if (atomic_add_fetch_32((x), 0) == old_) { \
       break; \
     } \
   }

From a28e9a9402fcb802fa7a301d76385fb083d6db3d Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 03:21:16 +0000
Subject: [PATCH 16/25] test

---
 src/client/src/tscServer.c | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 9425b32aab..7269173c8c 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -85,7 +85,7 @@ void tscUpdateMgmtIpList(SRpcIpSet *pIpSet) {
 }
 static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet, int32_t *vgId) {
   if (pVgroupInfo == NULL) { return;}
-  taosCorBeginRead(&pVgroupInfo->version);
+  //taosCorBeginRead(&pVgroupInfo->version);
   if (vgId) { 
     *vgId = pVgroupInfo->vgId; 
   }
@@ -96,7 +96,7 @@ static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pI
     strncpy(pIpSet->fqdn[i], pVgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
     pIpSet->port[i] = pVgroupInfo->ipAddr[i].port;
   }
-  taosCorEndRead(&pVgroupInfo->version);
+  //taosCorEndRead(&pVgroupInfo->version);
 }
 
 static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
@@ -105,7 +105,7 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
   if (pTableMetaInfo == NULL || pTableMetaInfo->pTableMeta == NULL) { return;}
   SCMVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->vgroupInfo;
 
-  taosCorBeginWrite(&pVgroupInfo->version);
+  //taosCorBeginWrite(&pVgroupInfo->version);
   //TODO(dengyihao), dont care vgid 
   pVgroupInfo->inUse = pIpSet->inUse;
   pVgroupInfo->numOfIps = pIpSet->numOfIps;
@@ -113,7 +113,7 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
     strncpy(pVgroupInfo->ipAddr[i].fqdn, pIpSet->fqdn[i], TSDB_FQDN_LEN);
     pVgroupInfo->ipAddr[i].port = pIpSet->port[i];
   }
-  taosCorEndWrite(&pVgroupInfo->version);
+  //taosCorEndWrite(&pVgroupInfo->version);
 }
 void tscPrintMgmtIp() {
   SRpcIpSet dump;

From 3c4e620f31257219d3871857f8c95102f87efecc Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 05:41:19 +0000
Subject: [PATCH 17/25] bugfix

---
 src/client/inc/tsclient.h      | 10 ++++++-
 src/client/src/tscSchemaUtil.c | 16 ++++++++---
 src/client/src/tscServer.c     | 49 ++++++++++++++++++++--------------
 src/inc/taosmsg.h              |  2 --
 4 files changed, 50 insertions(+), 27 deletions(-)

diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index e1180c4143..3a420567a9 100644
--- a/src/client/inc/tsclient.h
+++ b/src/client/inc/tsclient.h
@@ -52,12 +52,20 @@ typedef struct STableComInfo {
   int32_t rowSize;
 } STableComInfo;
 
+typedef struct SCMCorVgroupInfo {
+  int32_t version;
+  int8_t inUse;
+  int8_t  numOfIps;
+  SIpAddr ipAddr[TSDB_MAX_REPLICA];
+} SCMCorVgroupInfo;
+
 typedef struct STableMeta {
   STableComInfo tableInfo;
   uint8_t       tableType;
   int16_t       sversion;
   int16_t       tversion;
-  SCMVgroupInfo vgroupInfo;
+  SCMVgroupInfo  vgroupInfo;
+  SCMCorVgroupInfo  corVgroupInfo;
   int32_t       sid;       // the index of one table in a virtual node
   uint64_t      uid;       // unique id of a table
   SSchema       schema[];  // if the table is TSDB_CHILD_TABLE, schema is acquired by super table meta info
diff --git a/src/client/src/tscSchemaUtil.c b/src/client/src/tscSchemaUtil.c
index b391a5760e..52342b3650 100644
--- a/src/client/src/tscSchemaUtil.c
+++ b/src/client/src/tscSchemaUtil.c
@@ -140,7 +140,15 @@ struct SSchema tscGetTbnameColumnSchema() {
   strcpy(s.name, TSQL_TBNAME_L);
   return s;
 }
-
+static void tscInitCorVgroupInfo(SCMCorVgroupInfo *corVgroupInfo, SCMVgroupInfo *vgroupInfo) {
+  corVgroupInfo->version = 0;
+  corVgroupInfo->inUse = 0;
+  corVgroupInfo->numOfIps = vgroupInfo->numOfIps;
+  for (int32_t i = 0; i < corVgroupInfo->numOfIps; i++) {
+    strncpy(corVgroupInfo->ipAddr[i].fqdn, vgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
+    corVgroupInfo->ipAddr[i].port = vgroupInfo->ipAddr[i].port;
+  }
+}
 STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size) {
   assert(pTableMetaMsg != NULL);
   
@@ -157,9 +165,9 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
   pTableMeta->sid = pTableMetaMsg->sid;
   pTableMeta->uid = pTableMetaMsg->uid;
   pTableMeta->vgroupInfo = pTableMetaMsg->vgroup;
-  //init version here
-  pTableMeta->vgroupInfo.version = 0; 
-  
+
+  tscInitCorVgroupInfo(&pTableMeta->corVgroupInfo, &pTableMeta->vgroupInfo);
+
   pTableMeta->sversion = pTableMetaMsg->sversion;
   pTableMeta->tversion = pTableMetaMsg->tversion;
   
diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 7269173c8c..cb97307b04 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -45,6 +45,20 @@ void tscSaveSubscriptionProgress(void* sub);
 
 static int32_t minMsgSize() { return tsRpcHeadSize + 100; }
 
+static void tscSetDnodeIpList(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
+  SRpcIpSet* pIpList = &pSql->ipList;
+  pIpList->inUse    = 0;
+  if (pVgroupInfo == NULL) {
+    pIpList->numOfIps = 0;
+    return;
+  }
+
+  pIpList->numOfIps = pVgroupInfo->numOfIps;
+  for(int32_t i = 0; i < pVgroupInfo->numOfIps; ++i) {
+    strcpy(pIpList->fqdn[i], pVgroupInfo->ipAddr[i].fqdn);
+    pIpList->port[i] = pVgroupInfo->ipAddr[i].port;
+  }
+}
 void tscIpSetCopy(SRpcIpSet *dst, SRpcIpSet *src) {
   dst->numOfIps = src->numOfIps;
   dst->inUse = src->inUse;
@@ -83,12 +97,9 @@ void tscUpdateMgmtIpList(SRpcIpSet *pIpSet) {
   tscIpSetCopy(mgmtIpSet, pIpSet);
   taosCorEndWrite(&tscMgmtIpSet.version);
 }
-static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet, int32_t *vgId) {
+static void tscDumpIpSetFromVgroupInfo(SCMCorVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet) {
   if (pVgroupInfo == NULL) { return;}
-  //taosCorBeginRead(&pVgroupInfo->version);
-  if (vgId) { 
-    *vgId = pVgroupInfo->vgId; 
-  }
+  taosCorBeginRead(&pVgroupInfo->version);
   int8_t inUse = pVgroupInfo->inUse;
   pIpSet->inUse = (inUse >= 0 && inUse < TSDB_MAX_REPLICA) ? inUse: 0; 
   pIpSet->numOfIps = pVgroupInfo->numOfIps;  
@@ -96,16 +107,16 @@ static void tscDumpIpSetFromVgroupInfo(SCMVgroupInfo *pVgroupInfo, SRpcIpSet *pI
     strncpy(pIpSet->fqdn[i], pVgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
     pIpSet->port[i] = pVgroupInfo->ipAddr[i].port;
   }
-  //taosCorEndRead(&pVgroupInfo->version);
+  taosCorEndRead(&pVgroupInfo->version);
 }
 
 static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
   SSqlCmd *pCmd = &pObj->cmd;
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
   if (pTableMetaInfo == NULL || pTableMetaInfo->pTableMeta == NULL) { return;}
-  SCMVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->vgroupInfo;
+  SCMCorVgroupInfo *pVgroupInfo = &pTableMetaInfo->pTableMeta->corVgroupInfo;
 
-  //taosCorBeginWrite(&pVgroupInfo->version);
+  taosCorBeginWrite(&pVgroupInfo->version);
   //TODO(dengyihao), dont care vgid 
   pVgroupInfo->inUse = pIpSet->inUse;
   pVgroupInfo->numOfIps = pIpSet->numOfIps;
@@ -113,7 +124,7 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
     strncpy(pVgroupInfo->ipAddr[i].fqdn, pIpSet->fqdn[i], TSDB_FQDN_LEN);
     pVgroupInfo->ipAddr[i].port = pIpSet->port[i];
   }
-  //taosCorEndWrite(&pVgroupInfo->version);
+  taosCorEndWrite(&pVgroupInfo->version);
 }
 void tscPrintMgmtIp() {
   SRpcIpSet dump;
@@ -577,7 +588,7 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
 
   // pSql->cmd.payloadLen is set during copying data into payload
   pSql->cmd.msgType = TSDB_MSG_TYPE_SUBMIT;
-  tscDumpIpSetFromVgroupInfo(&pTableMeta->vgroupInfo, &pSql->ipList, NULL);
+  tscDumpIpSetFromVgroupInfo(&pTableMeta->corVgroupInfo, &pSql->ipList);
   
   tscDebug("%p build submit msg, vgId:%d numOfTables:%d numberOfIP:%d", pSql, vgId, pSql->cmd.numOfTablesInSubmit,
       pSql->ipList.numOfIps);
@@ -619,9 +630,10 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     } else {
       pVgroupInfo = &pTableMeta->vgroupInfo;
     }
-    int32_t vgId = 0;
-    tscDumpIpSetFromVgroupInfo(pVgroupInfo, &pSql->ipList, &vgId);
-    pQueryMsg->head.vgId = htonl(vgId);
+    tscSetDnodeIpList(pSql, pVgroupInfo);
+    if (pVgroupInfo != NULL) {
+      pQueryMsg->head.vgId = htonl(pVgroupInfo->vgId);
+    } 
 
     STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
     pTableIdInfo->tid = htonl(pTableMeta->sid);
@@ -639,10 +651,9 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
 
     SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, index);
     
-    // set the vgroup info
-    int32_t vgId = 0;
-    tscDumpIpSetFromVgroupInfo(&pTableIdList->vgInfo, &pSql->ipList, &vgId);
-    pQueryMsg->head.vgId = htonl(vgId);
+    // set the vgroup info 
+    tscSetDnodeIpList(pSql, &pTableIdList->vgInfo);
+    pQueryMsg->head.vgId = htonl(pTableIdList->vgInfo.vgId);
     
     int32_t numOfTables = taosArrayGetSize(pTableIdList->itemList);
     pQueryMsg->numOfTables = htonl(numOfTables);  // set the number of tables
@@ -1374,7 +1385,7 @@ int tscBuildUpdateTagMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
   SQueryInfo *    pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
 
-  tscDumpIpSetFromVgroupInfo(&pTableMetaInfo->pTableMeta->vgroupInfo, &pSql->ipList, NULL);
+  tscDumpIpSetFromVgroupInfo(&pTableMetaInfo->pTableMeta->corVgroupInfo, &pSql->ipList);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -1900,8 +1911,6 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
     for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) {
       //just init, no need to lock
       SCMVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
-      pVgroups->version = 0;
-      pVgroups->inUse = 0;
       pVgroups->vgId = htonl(pVgroups->vgId);
       assert(pVgroups->numOfIps >= 1);
       for (int32_t k = 0; k < pVgroups->numOfIps; ++k) {
diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h
index fec3d77d8e..b7afaf1e06 100644
--- a/src/inc/taosmsg.h
+++ b/src/inc/taosmsg.h
@@ -647,9 +647,7 @@ typedef struct SCMSTableVgroupMsg {
 } SCMSTableVgroupMsg, SCMSTableVgroupRspMsg;
 
 typedef struct {
-  int32_t   version;
   int32_t   vgId;
-  int8_t    inUse;
   int8_t    numOfIps;
   SIpAddr   ipAddr[TSDB_MAX_REPLICA];
 } SCMVgroupInfo;

From 246ea1df248e8721bad64f40fd57ed0ec0db263c Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 18 Jul 2020 10:31:31 +0000
Subject: [PATCH 18/25] modify from ip to ep

---
 src/client/inc/tsclient.h      |  4 +-
 src/client/src/tscSchemaUtil.c |  8 +--
 src/client/src/tscServer.c     | 95 +++++++++++++++-------------------
 src/client/src/tscSql.c        |  2 +-
 src/client/src/tscUtil.c       |  2 +-
 src/inc/trpc.h                 |  6 +--
 6 files changed, 53 insertions(+), 64 deletions(-)

diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h
index 83f9165d35..b5455ed1fb 100644
--- a/src/client/inc/tsclient.h
+++ b/src/client/inc/tsclient.h
@@ -55,8 +55,8 @@ typedef struct STableComInfo {
 typedef struct SCMCorVgroupInfo {
   int32_t version;
   int8_t inUse;
-  int8_t  numOfIps;
-  SIpAddr ipAddr[TSDB_MAX_REPLICA];
+  int8_t  numOfEps;
+  SEpAddr epAddr[TSDB_MAX_REPLICA];
 } SCMCorVgroupInfo;
 
 typedef struct STableMeta {
diff --git a/src/client/src/tscSchemaUtil.c b/src/client/src/tscSchemaUtil.c
index 52342b3650..9b8f48b109 100644
--- a/src/client/src/tscSchemaUtil.c
+++ b/src/client/src/tscSchemaUtil.c
@@ -143,10 +143,10 @@ struct SSchema tscGetTbnameColumnSchema() {
 static void tscInitCorVgroupInfo(SCMCorVgroupInfo *corVgroupInfo, SCMVgroupInfo *vgroupInfo) {
   corVgroupInfo->version = 0;
   corVgroupInfo->inUse = 0;
-  corVgroupInfo->numOfIps = vgroupInfo->numOfIps;
-  for (int32_t i = 0; i < corVgroupInfo->numOfIps; i++) {
-    strncpy(corVgroupInfo->ipAddr[i].fqdn, vgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
-    corVgroupInfo->ipAddr[i].port = vgroupInfo->ipAddr[i].port;
+  corVgroupInfo->numOfEps = vgroupInfo->numOfEps;
+  for (int32_t i = 0; i < corVgroupInfo->numOfEps; i++) {
+    strncpy(corVgroupInfo->epAddr[i].fqdn, vgroupInfo->epAddr[i].fqdn, TSDB_FQDN_LEN);
+    corVgroupInfo->epAddr[i].port = vgroupInfo->epAddr[i].port;
   }
 }
 STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size) {
diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index 3d78063bed..d9922b8718 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -53,64 +53,53 @@ static void tscSetDnodeEpSet(SSqlObj* pSql, SCMVgroupInfo* pVgroupInfo) {
     return;
   }
 
-  pEpList->numOfEps = pVgroupInfo->numOfEps;
+  pEpSet->numOfEps = pVgroupInfo->numOfEps;
   for(int32_t i = 0; i < pVgroupInfo->numOfEps; ++i) {
-    strcpy(pEpList->fqdn[i], pVgroupInfo->epAddr[i].fqdn);
-    pEpList->port[i] = pVgroupInfo->epAddr[i].port;
+    strcpy(pEpSet->fqdn[i], pVgroupInfo->epAddr[i].fqdn);
+    pEpSet->port[i] = pVgroupInfo->epAddr[i].port;
   }
 }
-void tscIpSetCopy(SRpcIpSet *dst, SRpcIpSet *src) {
-  dst->numOfIps = src->numOfIps;
-  dst->inUse = src->inUse;
-  for (int32_t i = 0; i < src->numOfIps; ++i) {
-     dst->port[i] = src->port[i];
-     strncpy(dst->fqdn[i], src->fqdn[i], TSDB_FQDN_LEN);
-  }
-}
-static void tscDumpMgmtIpSet(SRpcIpSet *ipSet) {
-  taosCorBeginRead(&tscMgmtIpSet.version);
-  SRpcIpSet*  src = &tscMgmtIpSet.ipSet;
-  tscIpSetCopy(ipSet, src); 
-  taosCorEndRead(&tscMgmtIpSet.version);
+static void tscDumpMgmtEpSet(SRpcEpSet *epSet) {
+  taosCorBeginRead(&tscMgmtEpSet.version);
+  *epSet = tscMgmtEpSet.epSet;
+  taosCorEndRead(&tscMgmtEpSet.version);
 }  
-static void tscIpSetHtons(SRpcIpSet *s) {
-   for (int32_t i = 0; i < s->numOfIps; i++) {
+static void tscEpSetHtons(SRpcEpSet *s) {
+   for (int32_t i = 0; i < s->numOfEps; i++) {
       s->port[i] = htons(s->port[i]);    
    }
 } 
-bool tscIpSetIsEqual(SRpcIpSet *s1, SRpcIpSet *s2) {
-   if (s1->numOfIps != s2->numOfIps || s1->inUse != s2->inUse) {
+bool tscEpSetIsEqual(SRpcEpSet *s1, SRpcEpSet *s2) {
+   if (s1->numOfEps != s2->numOfEps || s1->inUse != s2->inUse) {
      return false;
    } 
-   for (int32_t i = 0; i < s1->numOfIps; i++) {
+   for (int32_t i = 0; i < s1->numOfEps; i++) {
      if (s1->port[i] != s2->port[i] 
         || strncmp(s1->fqdn[i], s2->fqdn[i], TSDB_FQDN_LEN) != 0)
         return false;
    }
    return true;
 }
-void tscUpdateMgmtIpList(SRpcIpSet *pIpSet) {
+void tscUpdateMgmtEpSet(SRpcEpSet *pEpSet) {
   // no need to update if equal
-  taosCorBeginWrite(&tscMgmtIpSet.version);
-  // or copy directly,  tscMgmtIpSet.ipSet = *pIpSet
-  SRpcIpSet *mgmtIpSet = &tscMgmtIpSet.ipSet;
-  tscIpSetCopy(mgmtIpSet, pIpSet);
-  taosCorEndWrite(&tscMgmtIpSet.version);
+  taosCorBeginWrite(&tscMgmtEpSet.version);
+  tscMgmtEpSet.epSet = *pEpSet;
+  taosCorEndWrite(&tscMgmtEpSet.version);
 }
-static void tscDumpIpSetFromVgroupInfo(SCMCorVgroupInfo *pVgroupInfo, SRpcIpSet *pIpSet) {
+static void tscDumpEpSetFromVgroupInfo(SCMCorVgroupInfo *pVgroupInfo, SRpcEpSet *pEpSet) {
   if (pVgroupInfo == NULL) { return;}
   taosCorBeginRead(&pVgroupInfo->version);
   int8_t inUse = pVgroupInfo->inUse;
-  pIpSet->inUse = (inUse >= 0 && inUse < TSDB_MAX_REPLICA) ? inUse: 0; 
-  pIpSet->numOfIps = pVgroupInfo->numOfIps;  
-  for (int32_t i = 0; i < pVgroupInfo->numOfIps; ++i) {
-    strncpy(pIpSet->fqdn[i], pVgroupInfo->ipAddr[i].fqdn, TSDB_FQDN_LEN);
-    pIpSet->port[i] = pVgroupInfo->ipAddr[i].port;
+  pEpSet->inUse = (inUse >= 0 && inUse < TSDB_MAX_REPLICA) ? inUse: 0; 
+  pEpSet->numOfEps = pVgroupInfo->numOfEps;  
+  for (int32_t i = 0; i < pVgroupInfo->numOfEps; ++i) {
+    strncpy(pEpSet->fqdn[i], pVgroupInfo->epAddr[i].fqdn, TSDB_FQDN_LEN);
+    pEpSet->port[i] = pVgroupInfo->epAddr[i].port;
   }
   taosCorEndRead(&pVgroupInfo->version);
 }
 
-static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcIpSet *pIpSet) {
+static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcEpSet *pEpSet) {
   SSqlCmd *pCmd = &pObj->cmd;
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
   if (pTableMetaInfo == NULL || pTableMetaInfo->pTableMeta == NULL) { return;}
@@ -130,7 +119,7 @@ void tscPrintMgmtEp() {
   SRpcEpSet dump;
   tscDumpMgmtEpSet(&dump);
   if (dump.numOfEps <= 0) {
-    tscError("invalid mnode EP list:%d", dump.numOfEPs);
+    tscError("invalid mnode EP list:%d", dump.numOfEps);
   } else {
     for (int i = 0; i < dump.numOfEps; ++i) {
       tscDebug("mnode index:%d %s:%d", i, dump.fqdn[i], dump.port[i]);
@@ -166,10 +155,10 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
 
   if (code == 0) {
     SCMHeartBeatRsp *pRsp = (SCMHeartBeatRsp *)pRes->pRsp;
-    SRpcEpSet *      pEpList = &pRsp->epList;
-    if (pEpList->numOfEps > 0) {
-      tscEpSetHtons(pEpList);
-      tscUpdateMgmtEpList(pEpList);
+    SRpcEpSet *      epSet = &pRsp->epSet;
+    if (epSet->numOfEps > 0) {
+      tscEpSetHtons(epSet);
+      tscUpdateMgmtEpSet(epSet);
     } 
 
     pSql->pTscObj->connId = htonl(pRsp->connId);
@@ -242,7 +231,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
 
   // set the mgmt ip list
   if (pSql->cmd.command >= TSDB_SQL_MGMT) {
-    tscDumpMgmtEpSet(&pSql->epList);
+    tscDumpMgmtEpSet(&pSql->epSet);
   }
 
   memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
@@ -296,11 +285,11 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
   if (pEpSet) { 
     //SRpcEpSet dump; 
     tscEpSetHtons(pEpSet); 
-    if (tscEpSetIsEqual(&pSql->epList, pEpSet)) {
+    if (tscEpSetIsEqual(&pSql->epSet, pEpSet)) {
       if(pCmd->command < TSDB_SQL_MGMT)  { 
         tscUpdateVgroupInfo(pSql, pEpSet); 
       } else {
-        tscUpdateMgmtEpList(pEpSet);
+        tscUpdateMgmtEpSet(pEpSet);
     }
     }
   }
@@ -589,7 +578,7 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
 
   // pSql->cmd.payloadLen is set during copying data into payload
   pSql->cmd.msgType = TSDB_MSG_TYPE_SUBMIT;
-  tscDumpEpSetFromVgroupInfo(&pTableMeta->corVgroupInfo, &pSql->epList);
+  tscDumpEpSetFromVgroupInfo(&pTableMeta->corVgroupInfo, &pSql->epSet);
 
   tscDebug("%p build submit msg, vgId:%d numOfTables:%d numberOfEP:%d", pSql, vgId, pSql->cmd.numOfTablesInSubmit,
       pSql->epSet.numOfEps);
@@ -631,7 +620,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     } else {
       pVgroupInfo = &pTableMeta->vgroupInfo;
     }
-    tscSetDnodeEpList(pSql, pVgroupInfo);
+    tscSetDnodeEpSet(pSql, pVgroupInfo);
 
     if (pVgroupInfo != NULL) {
       pQueryMsg->head.vgId = htonl(pVgroupInfo->vgId);
@@ -654,7 +643,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
     SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, index);
 
     // set the vgroup info 
-    tscSetDnodeEpList(pSql, &pTableIdList->vgInfo);
+    tscSetDnodeEpSet(pSql, &pTableIdList->vgInfo);
     pQueryMsg->head.vgId = htonl(pTableIdList->vgInfo.vgId);
     
     int32_t numOfTables = taosArrayGetSize(pTableIdList->itemList);
@@ -1200,11 +1189,11 @@ int32_t tscBuildShowMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
       pShowMsg->payloadLen = htons(pPattern->n);
     }
   } else {
-    SSQLToken *pIpAddr = &pShowInfo->prefix;
-    assert(pIpAddr->n > 0 && pIpAddr->type > 0);
+    SSQLToken *pEpAddr = &pShowInfo->prefix;
+    assert(pEpAddr->n > 0 && pEpAddr->type > 0);
 
-    strncpy(pShowMsg->payload, pIpAddr->z, pIpAddr->n);
-    pShowMsg->payloadLen = htons(pIpAddr->n);
+    strncpy(pShowMsg->payload, pEpAddr->z, pEpAddr->n);
+    pShowMsg->payloadLen = htons(pEpAddr->n);
   }
 
   pCmd->payloadLen = sizeof(SCMShowMsg) + pShowMsg->payloadLen;
@@ -1387,7 +1376,7 @@ int tscBuildUpdateTagMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
   SQueryInfo *    pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
 
-  tscDumpEpSetFromVgroupInfo(&pTableMetaInfo->pTableMeta->corVgroupInfo, &pSql->epList);
+  tscDumpEpSetFromVgroupInfo(&pTableMetaInfo->pTableMeta->corVgroupInfo, &pSql->epSet);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -2011,9 +2000,9 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
   assert(len <= sizeof(pObj->db));
   tstrncpy(pObj->db, temp, sizeof(pObj->db));
   
-  if (pConnect->epList.numOfEps > 0) {
-    tscEpSetHtons(&pConnect->epList);
-    tscUpdateMgmtEpList(&pConnect->epList);
+  if (pConnect->epSet.numOfEps > 0) {
+    tscEpSetHtons(&pConnect->epSet);
+    tscUpdateMgmtEpSet(&pConnect->epSet);
   } 
 
   strcpy(pObj->sversion, pConnect->serverVersion);
diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c
index 59b506a454..5848b7b82f 100644
--- a/src/client/src/tscSql.c
+++ b/src/client/src/tscSql.c
@@ -62,7 +62,7 @@ SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
   }
 
   if (ip) {
-    if (tscSetMgmtEpListFromCfg(ip, NULL) < 0) return NULL;
+    if (tscSetMgmtEpSetFromCfg(ip, NULL) < 0) return NULL;
     if (port) tscMgmtEpSet.epSet.port[0] = port;
   } 
  
diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c
index c3cae58d8a..957bdeeb7f 100644
--- a/src/client/src/tscUtil.c
+++ b/src/client/src/tscUtil.c
@@ -2145,7 +2145,7 @@ char* strdup_throw(const char* str) {
   return p;
 }
 
-int tscSetMgmtEpListFromCfg(const char *first, const char *second) {
+int tscSetMgmtEpSetFromCfg(const char *first, const char *second) {
   // init mgmt ip set 
   tscMgmtEpSet.version = 0;
   SRpcEpSet *mgmtEpSet = &(tscMgmtEpSet.epSet);
diff --git a/src/inc/trpc.h b/src/inc/trpc.h
index 41ed98fd70..bdee917b5e 100644
--- a/src/inc/trpc.h
+++ b/src/inc/trpc.h
@@ -35,10 +35,10 @@ typedef struct SRpcEpSet {
   char      fqdn[TSDB_MAX_REPLICA][TSDB_FQDN_LEN];
 } SRpcEpSet;
 
-typedef struct SRpcCorIpSet {
+typedef struct SRpcCorEpSet {
   int32_t version; 
-  SRpcIpSet ipSet; 
-} SRpcCorIpSet;
+  SRpcEpSet epSet; 
+} SRpcCorEpSet;
 
 typedef struct SRpcConnInfo {
   uint32_t  clientIp;

From 4a6c170d96607251389741df7fbc8fc7ab5019ac Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Sun, 19 Jul 2020 17:53:06 +0800
Subject: [PATCH 19/25] change hash data in sdb

---
 src/mnode/src/mnodeSdb.c | 79 ++++++++++++++--------------------------
 1 file changed, 27 insertions(+), 52 deletions(-)

diff --git a/src/mnode/src/mnodeSdb.c b/src/mnode/src/mnodeSdb.c
index 0f72dbdec4..42ded7ed06 100644
--- a/src/mnode/src/mnodeSdb.c
+++ b/src/mnode/src/mnodeSdb.c
@@ -77,11 +77,6 @@ typedef struct {
   pthread_mutex_t mutex;
 } SSdbObject;
 
-typedef struct {
-  int32_t rowSize;
-  void *  row;
-} SSdbRow;
-
 typedef struct {
   pthread_t thread;
   int32_t   workerId;
@@ -419,32 +414,28 @@ void sdbDecRef(void *handle, void *pObj) {
   }
 }
 
-static SSdbRow *sdbGetRowMeta(SSdbTable *pTable, void *key) {
+static void *sdbGetRowMeta(SSdbTable *pTable, void *key) {
   if (pTable == NULL) return NULL;
 
   int32_t keySize = sizeof(int32_t);
   if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
     keySize = strlen((char *)key);
   }
-  
-  return taosHashGet(pTable->iHandle, key, keySize);
+
+  void **ppRow = (void **)taosHashGet(pTable->iHandle, key, keySize);
+  if (ppRow == NULL) return NULL;
+  return *ppRow;
 }
 
-static SSdbRow *sdbGetRowMetaFromObj(SSdbTable *pTable, void *key) {
+static void *sdbGetRowMetaFromObj(SSdbTable *pTable, void *key) {
   return sdbGetRowMeta(pTable, sdbGetObjKey(pTable, key));
 }
 
 void *sdbGetRow(void *handle, void *key) {
-  SSdbTable *pTable = (SSdbTable *)handle;
-  int32_t keySize = sizeof(int32_t);
-  if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
-    keySize = strlen((char *)key);
-  }
-  
-  SSdbRow *pMeta = taosHashGet(pTable->iHandle, key, keySize);
-  if (pMeta) {
-    sdbIncRef(pTable, pMeta->row);
-    return pMeta->row;
+  void *pRow = sdbGetRowMeta(handle, key);
+  if (pRow) {
+    sdbIncRef(handle, pRow);
+    return pRow;
   } else {
     return NULL;
   }
@@ -455,10 +446,6 @@ static void *sdbGetRowFromObj(SSdbTable *pTable, void *key) {
 }
 
 static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
-  SSdbRow rowMeta;
-  rowMeta.rowSize = pOper->rowSize;
-  rowMeta.row = pOper->pObj;
-
   void *  key = sdbGetObjKey(pTable, pOper->pObj);
   int32_t keySize = sizeof(int32_t);
 
@@ -466,7 +453,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
     keySize = strlen((char *)key);
   }
 
-  taosHashPut(pTable->iHandle, key, keySize, &rowMeta, sizeof(SSdbRow));
+  taosHashPut(pTable->iHandle, key, keySize, &pOper->pObj, sizeof(void **));
 
   sdbIncRef(pTable, pOper->pObj);
   atomic_add_fetch_32(&pTable->numOfRows, 1);
@@ -586,17 +573,17 @@ static int sdbWrite(void *param, void *data, int type) {
     code = (*pTable->decodeFp)(&oper);
     return sdbInsertHash(pTable, &oper);
   } else if (action == SDB_ACTION_DELETE) {
-    SSdbRow *rowMeta = sdbGetRowMeta(pTable, pHead->cont);
-    if (rowMeta == NULL || rowMeta->row == NULL) {
+    void *pRow = sdbGetRowMeta(pTable, pHead->cont);
+    if (pRow == NULL) {
       sdbError("table:%s, failed to get object:%s from wal while dispose delete action", pTable->tableName,
                pHead->cont);
       return TSDB_CODE_SUCCESS;
     }
-    SSdbOper oper = {.table = pTable, .pObj = rowMeta->row};
+    SSdbOper oper = {.table = pTable, .pObj = pRow};
     return sdbDeleteHash(pTable, &oper);
   } else if (action == SDB_ACTION_UPDATE) {
-    SSdbRow *rowMeta = sdbGetRowMeta(pTable, pHead->cont);
-    if (rowMeta == NULL || rowMeta->row == NULL) {
+    void *pRow = sdbGetRowMeta(pTable, pHead->cont);
+    if (pRow == NULL) {
       sdbError("table:%s, failed to get object:%s from wal while dispose update action", pTable->tableName,
                pHead->cont);
       return TSDB_CODE_SUCCESS;
@@ -675,18 +662,12 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
   if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
-  SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
-  if (pMeta == NULL) {
+  void *pRow = sdbGetRowMetaFromObj(pTable, pOper->pObj);
+  if (pRow == NULL) {
     sdbDebug("table:%s, record is not there, delete failed", pTable->tableName);
     return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
   }
 
-  void *pMetaRow = pMeta->row;
-  if (pMetaRow == NULL) {
-    sdbError("table:%s, record meta is null", pTable->tableName);
-    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
-  }
-
   sdbIncRef(pTable, pOper->pObj);
 
   int32_t code = sdbDeleteHash(pTable, pOper);
@@ -728,18 +709,12 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
   SSdbTable *pTable = (SSdbTable *)pOper->table;
   if (pTable == NULL) return TSDB_CODE_MND_SDB_INVALID_TABLE_TYPE;
 
-  SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
-  if (pMeta == NULL) {
+  void *pRow = sdbGetRowMetaFromObj(pTable, pOper->pObj);
+  if (pRow == NULL) {
     sdbDebug("table:%s, record is not there, update failed", pTable->tableName);
     return TSDB_CODE_MND_SDB_OBJ_NOT_THERE;
   }
 
-  void *pMetaRow = pMeta->row;
-  if (pMetaRow == NULL) {
-    sdbError("table:%s, record meta is null", pTable->tableName);
-    return TSDB_CODE_MND_SDB_INVAID_META_ROW;
-  }
-
   int32_t code = sdbUpdateHash(pTable, pOper);
   if (code != TSDB_CODE_SUCCESS) {
     sdbError("table:%s, failed to update hash", pTable->tableName);
@@ -789,14 +764,14 @@ void *sdbFetchRow(void *handle, void *pNode, void **ppRow) {
     return NULL;
   }
 
-  SSdbRow *pMeta = taosHashIterGet(pIter);
-  if (pMeta == NULL) {
+  void **ppMetaRow = taosHashIterGet(pIter);
+  if (ppMetaRow == NULL) {
     taosHashDestroyIter(pIter);
     return NULL;
   }
 
-  *ppRow = pMeta->row;
-  sdbIncRef(handle, pMeta->row);
+  *ppRow = *ppMetaRow;
+  sdbIncRef(handle, *ppMetaRow);
 
   return pIter;
 }
@@ -846,11 +821,11 @@ void sdbCloseTable(void *handle) {
 
   SHashMutableIterator *pIter = taosHashCreateIter(pTable->iHandle);
   while (taosHashIterNext(pIter)) {
-    SSdbRow *pMeta = taosHashIterGet(pIter);
-    if (pMeta == NULL) continue;
+    void **ppRow = taosHashIterGet(pIter);
+    if (ppRow == NULL) continue;
 
     SSdbOper oper = {
-      .pObj = pMeta->row,
+      .pObj = *ppRow,
       .table = pTable,
     };
 

From 19e2109614435838e028f9db70c780ae5ea8239f Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Sun, 19 Jul 2020 18:11:23 +0800
Subject: [PATCH 20/25] fix refcount error while drop stable

---
 src/mnode/src/mnodeTable.c | 41 +++++++++++++++++++-------------------
 1 file changed, 20 insertions(+), 21 deletions(-)

diff --git a/src/mnode/src/mnodeTable.c b/src/mnode/src/mnodeTable.c
index 75ed442cd4..7478d7cd78 100644
--- a/src/mnode/src/mnodeTable.c
+++ b/src/mnode/src/mnodeTable.c
@@ -107,42 +107,41 @@ static int32_t mnodeChildTableActionInsert(SSdbOper *pOper) {
   SVgObj *pVgroup = mnodeGetVgroup(pTable->vgId);
   if (pVgroup == NULL) {
     mError("ctable:%s, not in vgId:%d", pTable->info.tableId, pTable->vgId);
-    return TSDB_CODE_MND_VGROUP_NOT_EXIST;
   }
-  mnodeDecVgroupRef(pVgroup);
 
-  SDbObj *pDb = mnodeGetDb(pVgroup->dbName);
-  if (pDb == NULL) {
-    mError("ctable:%s, vgId:%d not in db:%s", pTable->info.tableId, pVgroup->vgId, pVgroup->dbName);
-    return TSDB_CODE_MND_INVALID_DB;
+  SDbObj *pDb = NULL;
+  if (pVgroup != NULL) {
+    pDb = mnodeGetDb(pVgroup->dbName);
+    if (pDb == NULL) {
+      mError("ctable:%s, vgId:%d not in db:%s", pTable->info.tableId, pVgroup->vgId, pVgroup->dbName);
+    }
   }
-  
-  if (pDb->status != TSDB_DB_STATUS_READY) {
-    mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
-    return TSDB_CODE_MND_DB_IN_DROPPING;
-  }
-  mnodeDecDbRef(pDb);
 
-  SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
-  if (pAcct == NULL) {
-    mError("ctable:%s, acct:%s not exists", pTable->info.tableId, pDb->acct);
-    return TSDB_CODE_MND_INVALID_ACCT;
+  SAcctObj *pAcct = NULL;
+  if (pDb != NULL) {
+    pAcct = mnodeGetAcct(pDb->acct);
+    if (pAcct == NULL) {
+      mError("ctable:%s, acct:%s not exists", pTable->info.tableId, pDb->acct);
+    }
   }
-  mnodeDecAcctRef(pAcct);
 
   if (pTable->info.type == TSDB_CHILD_TABLE) {
     // add ref
     pTable->superTable = mnodeGetSuperTableByUid(pTable->suid);
     mnodeAddTableIntoStable(pTable->superTable, pTable);
     grantAdd(TSDB_GRANT_TIMESERIES, pTable->superTable->numOfColumns - 1);
-    pAcct->acctInfo.numOfTimeSeries += (pTable->superTable->numOfColumns - 1);
+    if (pAcct) pAcct->acctInfo.numOfTimeSeries += (pTable->superTable->numOfColumns - 1);
   } else {
     grantAdd(TSDB_GRANT_TIMESERIES, pTable->numOfColumns - 1);
-    pAcct->acctInfo.numOfTimeSeries += (pTable->numOfColumns - 1);
+    if (pAcct) pAcct->acctInfo.numOfTimeSeries += (pTable->numOfColumns - 1);
   }
 
-  mnodeAddTableIntoDb(pDb);
-  mnodeAddTableIntoVgroup(pVgroup, pTable);
+  if (pDb) mnodeAddTableIntoDb(pDb);
+  if (pVgroup) mnodeAddTableIntoVgroup(pVgroup, pTable);
+
+  mnodeDecVgroupRef(pVgroup);
+  mnodeDecDbRef(pDb);
+  mnodeDecAcctRef(pAcct);
 
   return TSDB_CODE_SUCCESS;
 }

From b40c7018e3fe9b0b0525c4b2fdbb587179e89c5f Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Mon, 20 Jul 2020 09:28:13 +0800
Subject: [PATCH 21/25] print messages in red on error

---
 tests/examples/c/apitest.c | 20 ++++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)

diff --git a/tests/examples/c/apitest.c b/tests/examples/c/apitest.c
index bc628fb58f..759e16d1de 100644
--- a/tests/examples/c/apitest.c
+++ b/tests/examples/c/apitest.c
@@ -218,9 +218,9 @@ void verify_prepare(TAOS* taos) {
 
   int code = taos_errno(result);
   if (code != 0) {
-    printf("failed to create database, reason:%s\n", taos_errstr(result));
+    printf("\033[31mfailed to create database, reason:%s\033[0m\n", taos_errstr(result));
     taos_free_result(result);
-    exit(1);
+    return;
   }
   taos_free_result(result);
 
@@ -232,9 +232,9 @@ void verify_prepare(TAOS* taos) {
   result = taos_query(taos, sql);
   code = taos_errno(result);
   if (code != 0) {
-    printf("failed to create table, reason:%s\n", taos_errstr(result));
+    printf("\033[31mfailed to create table, reason:%s\033[0m\n", taos_errstr(result));
     taos_free_result(result);
-    exit(1);
+    return;
   }
   taos_free_result(result);
 
@@ -320,7 +320,7 @@ void verify_prepare(TAOS* taos) {
   sql = "insert into m1 values(?,?,?,?,?,?,?,?,?,?)";
   code = taos_stmt_prepare(stmt, sql, 0);
   if (code != 0){
-    printf("failed to execute taos_stmt_prepare. code:0x%x\n", code);
+    printf("\033[31mfailed to execute taos_stmt_prepare. code:0x%x\033[0m\n", code);
   }
   v.ts = 1591060628000;
   for (int i = 0; i < 10; ++i) {
@@ -343,8 +343,8 @@ void verify_prepare(TAOS* taos) {
     taos_stmt_add_batch(stmt);
   }
   if (taos_stmt_execute(stmt) != 0) {
-    printf("failed to execute insert statement.\n");
-    exit(1);
+    printf("\033[31mfailed to execute insert statement.\033[0m\n");
+    return;
   }
   taos_stmt_close(stmt);
 
@@ -355,8 +355,8 @@ void verify_prepare(TAOS* taos) {
   v.v2 = 15;
   taos_stmt_bind_param(stmt, params + 2);
   if (taos_stmt_execute(stmt) != 0) {
-    printf("failed to execute select statement.\n");
-    exit(1);
+    printf("\033[31mfailed to execute select statement.\033[0m\n");
+    return;
   }
 
   result = taos_stmt_use_result(stmt);
@@ -444,7 +444,7 @@ int main(int argc, char *argv[]) {
 
   TAOS* taos = taos_connect(host, user, passwd, "", 0);
   if (taos == NULL) {
-    printf("failed to connect to db, reason:%s\n", taos_errstr(taos));
+    printf("\033[31mfailed to connect to db, reason:%s\033[0m\n", taos_errstr(taos));
     exit(1);
   }
 

From 1067175746851366b658b0b2b590f77ff915642f Mon Sep 17 00:00:00 2001
From: Bomin Zhang <bmzhang@taosdata.com>
Date: Mon, 20 Jul 2020 11:15:32 +0800
Subject: [PATCH 22/25] fix compiler error

---
 src/client/src/tscSub.c   | 3 ++-
 tests/examples/c/makefile | 2 +-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c
index 73614fa0a5..526f380475 100644
--- a/src/client/src/tscSub.c
+++ b/src/client/src/tscSub.c
@@ -92,6 +92,7 @@ static void asyncCallback(void *param, TAOS_RES *tres, int code) {
 
 static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* sql) {
   int code = TSDB_CODE_SUCCESS, line = __LINE__;
+  SSqlObj* pSql = NULL;
 
   SSub* pSub = calloc(1, sizeof(SSub));
   if (pSub == NULL) {
@@ -113,7 +114,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
     goto fail;
   }
 
-  SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
+  pSql = calloc(1, sizeof(SSqlObj));
   if (pSql == NULL) {
     line = __LINE__;
     code = TSDB_CODE_TSC_OUT_OF_MEMORY;
diff --git a/tests/examples/c/makefile b/tests/examples/c/makefile
index 0a4b8ee9d2..f9653c9c96 100644
--- a/tests/examples/c/makefile
+++ b/tests/examples/c/makefile
@@ -4,7 +4,6 @@
 ROOT=./
 TARGET=exe
 LFLAGS = '-Wl,-rpath,/usr/local/taos/driver/' -ltaos -lpthread -lm -lrt
-#LFLAGS = '-Wl,-rpath,/home/zbm/project/td/debug/build/lib/' -L/home/zbm/project/td/debug/build/lib -ltaos -lpthread -lm -lrt
 CFLAGS = -O3 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion -Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX -msse4.2 -Wno-unused-function -D_M_X64 \
 	-I/usr/local/taos/include -std=gnu99
 
@@ -16,6 +15,7 @@ exe:
 	gcc $(CFLAGS) ./prepare.c -o $(ROOT)/prepare $(LFLAGS)
 	gcc $(CFLAGS) ./stream.c -o $(ROOT)/stream $(LFLAGS)
 	gcc $(CFLAGS) ./subscribe.c -o $(ROOT)subscribe $(LFLAGS)
+	gcc $(CFLAGS) ./apitest.c -o $(ROOT)apitest $(LFLAGS)
 
 clean:
 	rm $(ROOT)/asyncdemo

From caa5442005f28a0e4c22b538ff64bfc6ffa124a2 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sun, 19 Jul 2020 12:27:45 +0000
Subject: [PATCH 23/25] src/inc/taos.h

---
 src/inc/taos.h | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/src/inc/taos.h b/src/inc/taos.h
index d6f1883572..1d609bc7db 100644
--- a/src/inc/taos.h
+++ b/src/inc/taos.h
@@ -67,8 +67,6 @@ DLL_EXPORT void  taos_init();
 DLL_EXPORT void  taos_cleanup();
 DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
 DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_c(const char *ip, uint8_t ipLen, const char *user, uint8_t userLen,
-                 const char *pass, uint8_t passLen, const char *db, uint8_t dbLen, uint16_t port);
 DLL_EXPORT void  taos_close(TAOS *taos);
 
 typedef struct TAOS_BIND {
@@ -90,7 +88,6 @@ TAOS_RES * taos_stmt_use_result(TAOS_STMT *stmt);
 int        taos_stmt_close(TAOS_STMT *stmt);
 
 DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
-DLL_EXPORT TAOS_RES *taos_query_c(TAOS *taos, const char *sql, uint32_t sqlLen);
 DLL_EXPORT TAOS_ROW taos_fetch_row(TAOS_RES *res);
 DLL_EXPORT int taos_result_precision(TAOS_RES *res);  // get the time precision of result
 DLL_EXPORT void taos_free_result(TAOS_RES *res);

From ac1d0a266b39c5d8dfc2e17f78d0da51b0da67d7 Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sdsang@taosdata.com>
Date: Mon, 20 Jul 2020 11:36:12 +0800
Subject: [PATCH 24/25] fix query unittest compile error.

[TD-964]
---
 src/query/tests/resultBufferTest.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/query/tests/resultBufferTest.cpp b/src/query/tests/resultBufferTest.cpp
index 9270c6e458..aaea0836d1 100644
--- a/src/query/tests/resultBufferTest.cpp
+++ b/src/query/tests/resultBufferTest.cpp
@@ -22,7 +22,7 @@ void simpleTest() {
   ASSERT_EQ(getResBufSize(pResultBuf), 1000*16384L);
   
   SIDList list = getDataBufPagesIdList(pResultBuf, groupId);
-  ASSERT_EQ(list.size, 1);
+  ASSERT_EQ(taosArrayGetSize(list), 1);
   
   ASSERT_EQ(getNumOfResultBufGroupId(pResultBuf), 1);
   

From dfe2c4fc67756167c13cfebe0de4ca515e1514b8 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 20 Jul 2020 17:16:59 +0800
Subject: [PATCH 25/25] [td-225] fix memory leaks.

---
 src/query/inc/qResultbuf.h | 1 +
 src/query/src/qResultbuf.c | 4 +++-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/query/inc/qResultbuf.h b/src/query/inc/qResultbuf.h
index d76659cc91..8c8afb0957 100644
--- a/src/query/inc/qResultbuf.h
+++ b/src/query/inc/qResultbuf.h
@@ -42,6 +42,7 @@ typedef struct SDiskbasedResultBuf {
 
   void*     iBuf;                // inmemory buf
   void*     handle;              // for debug purpose
+  void*     emptyDummyIdList;    // dummy id list
 } SDiskbasedResultBuf;
 
 #define DEFAULT_INTERN_BUF_PAGE_SIZE (1024L)
diff --git a/src/query/src/qResultbuf.c b/src/query/src/qResultbuf.c
index 0eefb106b9..de59676e59 100644
--- a/src/query/src/qResultbuf.c
+++ b/src/query/src/qResultbuf.c
@@ -36,6 +36,7 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t nu
 
   pResBuf->fd = FD_INITIALIZER;
   pResBuf->pBuf = NULL;
+  pResBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
 
   qDebug("QInfo:%p create resBuf for output, page size:%d, initial pages:%d, %" PRId64 "bytes", handle,
       pResBuf->pageSize, pResBuf->numOfPages, pResBuf->totalBufSize);
@@ -173,7 +174,7 @@ int32_t getNumOfRowsPerPage(SDiskbasedResultBuf* pResultBuf) { return pResultBuf
 SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId) {
   int32_t slot = getGroupIndex(pResultBuf, groupId);
   if (slot < 0) {
-    return taosArrayInit(1, sizeof(int32_t));
+    return pResultBuf->emptyDummyIdList;
   } else {
     return taosArrayGetP(pResultBuf->list, slot);
   }
@@ -206,6 +207,7 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) {
   }
 
   taosArrayDestroy(pResultBuf->list);
+  taosArrayDestroy(pResultBuf->emptyDummyIdList);
   taosHashCleanup(pResultBuf->idsTable);
 
   tfree(pResultBuf->iBuf);