[td-225]

src/client/inc/tscUtil.h

@@ -247,6 +247,8 @@ void tscDoQuery(SSqlObj* pSql);
  * @param pPrevSql
  * @return
  */
+SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd);
+
 SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void* param, int32_t cmd, SSqlObj* pPrevSql);
 void     addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClauseIndex, int32_t tableIndex);
 

src/client/inc/tsclient.h

@@ -296,7 +296,7 @@ typedef struct STscObj {
 typedef struct SSqlObj {
   void            *signature;
   STscObj         *pTscObj;
-  void            *SRpcReqContext;
+  void            *pRpcCtx;
   void            (*fp)();
   void            (*fetchFp)();
   void            *param;
@@ -308,8 +308,7 @@ typedef struct SSqlObj {
   char             retry;
   char             maxRetry;
   SRpcIpSet        ipList;
-  char             freed : 4;
-  char             listed : 4;
+  char             listed;
   tsem_t           rspSem;
   SSqlCmd          cmd;
   SSqlRes          res;
@@ -349,7 +348,7 @@ typedef struct SSqlStream {
 int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn);
 void    tscInitMsgsFp();
 
-int tsParseSql(SSqlObj *pSql, bool initialParse);
+int tsParseSql(SSqlObj *pSql, bool initial);
 
 void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet);
 int  tscProcessSql(SSqlObj *pSql);

src/client/src/tscServer.c

@@ -197,29 +197,34 @@ int tscSendMsgToServer(SSqlObj *pSql) {
       .code    = 0
   };
 
-  pSql->SRpcReqContext = rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg);
+  pSql->pRpcCtx = rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg);
   return TSDB_CODE_SUCCESS;
 }
 
 void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
   SSqlObj *pSql = (SSqlObj *)rpcMsg->handle;
-  if (pSql == NULL) {
+  if (pSql == NULL || pSql->signature != pSql) {
     tscError("%p sql is already released", pSql->signature);
     return;
   }
 
-  if (pSql->signature != pSql) {
-    tscError("%p sql is already released, signature:%p", pSql, pSql->signature);
-    return;
-  }
-
+  STscObj *pObj = pSql->pTscObj;
   SSqlRes *pRes = &pSql->res;
   SSqlCmd *pCmd = &pSql->cmd;
-  STscObj *pObj = pSql->pTscObj;
 
-  if (pObj->signature != pObj || pSql->freed == 1) {
-    tscTrace("%p sqlObj needs to be released or DB connection is closed, freed:%d pObj:%p signature:%p", pSql, pSql->freed,
+  if (pObj->signature != pObj) {
+    tscTrace("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature);
+
+    tscFreeSqlObj(pSql);
+    rpcFreeCont(rpcMsg->pCont);
+    return;
+  }
+
+  SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
+  if (pQueryInfo != NULL && pQueryInfo->type == TSDB_QUERY_TYPE_FREE_RESOURCE) {
+    tscTrace("%p sqlObj needs to be released or DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command,
              pObj, pObj->signature);
+
     tscFreeSqlObj(pSql);
     rpcFreeCont(rpcMsg->pCont);
     return;
@@ -421,7 +426,7 @@ void tscKillSTableQuery(SSqlObj *pSql) {
      * sub-queries not correctly released and master sql object of super table query reaches an abnormal state.
      */
     pSql->pSubs[i]->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
-    rpcCancelRequest(pSql->pSubs[i]->SRpcReqContext);
+    rpcCancelRequest(pSql->pSubs[i]->pRpcCtx);
   }
 
   /*

src/client/src/tscSql.c

@@ -489,7 +489,6 @@ static bool tscFreeQhandleInVnode(SSqlObj* pSql) {
 
     pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
     tscTrace("%p send msg to dnode to free qhandle ASAP, command:%s", pSql, sqlCmd[pCmd->command]);
-    pSql->freed = 1;
     tscProcessSql(pSql);
 
     // in case of sync model query, waits for response and then goes on
@@ -631,7 +630,7 @@ void taos_stop_query(TAOS_RES *res) {
     return;
   }
 
-  rpcCancelRequest(pSql->SRpcReqContext);
+  rpcCancelRequest(pSql->pRpcCtx);
   tscTrace("%p query is cancelled", res);
 }
 

src/client/src/tscSubquery.c

@@ -1853,11 +1853,11 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
   SSubqueryState* pState = pSupporter->pState;
 
   // record the total inserted rows
-  if (numOfRows > 0 && tres != pParentObj) {
-      pParentObj->res.numOfRows += numOfRows;
+  if (numOfRows > 0) {
+    pParentObj->res.numOfRows += numOfRows;
   }
 
-  if (taos_errno(tres) != 0) {
+  if (taos_errno(tres) != TSDB_CODE_SUCCESS) {
     SSqlObj* pSql = (SSqlObj*) tres;
     assert(pSql != NULL && pSql->res.code == numOfRows);
     
@@ -1865,13 +1865,9 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
   }
 
   // it is not the initial sqlObj, free it
-  if (tres != pParentObj) {
-    taos_free_result(tres);
-  } else {
-    assert(pParentObj->pSubs[0] == tres);
-  }
-
+  taos_free_result(tres);
   tfree(pSupporter);
+
   if (atomic_sub_fetch_32(&pState->numOfRemain, 1) > 0) {
     return;
   }
@@ -1904,30 +1900,14 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
   pState->numOfRemain = pSql->numOfSubs;
  
   pRes->code = TSDB_CODE_SUCCESS;
+  int32_t numOfSub = 0;
 
-  SInsertSupporter* pSupporter = calloc(1, sizeof(SInsertSupporter));
-  pSupporter->pSql   = pSql;
-  pSupporter->pState = pState;
-
-  pSql->fp       = multiVnodeInsertFinalize;
-  pSql->param    = pSupporter;
-  pSql->pSubs[0] = pSql;  // the first sub insert points back to itself
-  tscTrace("%p sub:%p create subObj success. orderOfSub:%d", pSql, pSql, 0);
-
-  int32_t numOfSub = 1;
-  int32_t code = tscCopyDataBlockToPayload(pSql, pDataBlocks->pData[0]);
-  if (code != TSDB_CODE_SUCCESS) {
-    tscTrace("%p prepare submit data block failed in async insertion, vnodeIdx:%d, total:%d, code:%d", pSql, 0,
-             pDataBlocks->nSize, code);
-    goto _error;
-  }
-
-  for (; numOfSub < pSql->numOfSubs; ++numOfSub) {
-    SInsertSupporter* pSupporter1 = calloc(1, sizeof(SInsertSupporter));
-    pSupporter1->pSql = pSql;
-    pSupporter1->pState = pState;
+  while(numOfSub < pSql->numOfSubs) {
+    SInsertSupporter* pSupporter = calloc(1, sizeof(SInsertSupporter));
+    pSupporter->pSql = pSql;
+    pSupporter->pState = pState;
     
-    SSqlObj *pNew = createSubqueryObj(pSql, 0, multiVnodeInsertFinalize, pSupporter1, TSDB_SQL_INSERT, NULL);
+    SSqlObj *pNew = createSimpleSubObj(pSql, multiVnodeInsertFinalize, pSupporter, TSDB_SQL_INSERT);//createSubqueryObj(pSql, 0, multiVnodeInsertFinalize, pSupporter1, TSDB_SQL_INSERT, NULL);
     if (pNew == NULL) {
       tscError("%p failed to malloc buffer for subObj, orderOfSub:%d, reason:%s", pSql, numOfSub, strerror(errno));
       goto _error;
@@ -1940,13 +1920,13 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
     pNew->fetchFp = pNew->fp;
     pSql->pSubs[numOfSub] = pNew;
 
-    code = tscCopyDataBlockToPayload(pNew, pDataBlocks->pData[numOfSub]);
-    if (code != TSDB_CODE_SUCCESS) {
-      tscTrace("%p prepare submit data block failed in async insertion, vnodeIdx:%d, total:%d, code:%d", pSql, numOfSub,
-               pDataBlocks->nSize, code);
-      goto _error;
-    } else {
+    pRes->code = tscCopyDataBlockToPayload(pNew, pDataBlocks->pData[numOfSub++]);
+    if (pRes->code == TSDB_CODE_SUCCESS) {
       tscTrace("%p sub:%p create subObj success. orderOfSub:%d", pSql, pNew, numOfSub);
+    } else {
+      tscTrace("%p prepare submit data block failed in async insertion, vnodeIdx:%d, total:%d, code:%s", pSql, numOfSub,
+               pDataBlocks->nSize, tstrerror(pRes->code));
+      goto _error;
     }
   }
   
@@ -1966,18 +1946,12 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
   return TSDB_CODE_SUCCESS;
 
   _error:
-  // restore the udf fp
-  pSql->fp = pSql->fetchFp;
-  pSql->pSubs[0] = NULL;
-
-  tfree(pState);
-  tfree(pSql->param);
-
-  for(int32_t j = 1; j < numOfSub; ++j) {
+  for(int32_t j = 0; j < numOfSub; ++j) {
     tfree(pSql->pSubs[j]->param);
     taos_free_result(pSql->pSubs[j]);
   }
 
+  tfree(pState);
   return TSDB_CODE_TSC_OUT_OF_MEMORY;
 }
 

src/client/src/tscUtil.c

@@ -364,7 +364,6 @@ void tscPartiallyFreeSqlObj(SSqlObj* pSql) {
   tscFreeSqlResult(pSql);
   
   tfree(pSql->pSubs);
-  pSql->freed = 0;
   pSql->numOfSubs = 0;
   
   tscResetSqlCmdObj(pCmd);
@@ -1653,6 +1652,38 @@ void tscResetForNextRetrieve(SSqlRes* pRes) {
   pRes->numOfRows = 0;
 }
 
+SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd) {
+  SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj));
+  if (pNew == NULL) {
+    tscError("%p new subquery failed, tableIndex:%d", pSql, 0);
+    return NULL;
+  }
+
+  pNew->pTscObj = pSql->pTscObj;
+  pNew->signature = pNew;
+
+  SSqlCmd* pCmd = &pNew->cmd;
+  pCmd->command = cmd;
+
+  if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
+    tscFreeSqlObj(pNew);
+    return NULL;
+  }
+
+  pNew->fp = fp;
+  pNew->param = param;
+  pNew->maxRetry = TSDB_MAX_REPLICA_NUM;
+
+  SQueryInfo* pQueryInfo = NULL;
+  tscGetQueryInfoDetailSafely(pCmd, 0, &pQueryInfo);
+
+  assert(pSql->cmd.clauseIndex == 0);
+  STableMetaInfo* pMasterTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, pSql->cmd.clauseIndex, 0);
+
+  tscAddTableMetaInfo(pQueryInfo, pMasterTableMetaInfo->name, NULL, NULL, NULL);
+  return pNew;
+}
+
 SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void* param, int32_t cmd, SSqlObj* pPrevSql) {
   SSqlCmd* pCmd = &pSql->cmd;
   SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj));
@@ -1666,14 +1697,6 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
   pNew->pTscObj = pSql->pTscObj;
   pNew->signature = pNew;
 
-  pNew->sqlstr = strdup(pSql->sqlstr);
-  if (pNew->sqlstr == NULL) {
-    tscError("%p new subquery failed, tableIndex:%d, vgroupIndex:%d", pSql, tableIndex, pTableMetaInfo->vgroupIndex);
-
-    free(pNew);
-    return NULL;
-  }
-
   SSqlCmd* pnCmd = &pNew->cmd;
   memcpy(pnCmd, pCmd, sizeof(SSqlCmd));
   
@@ -1795,6 +1818,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
   if (pPrevSql == NULL) {
     STableMeta* pTableMeta = taosCacheAcquireByData(tscCacheHandle, pTableMetaInfo->pTableMeta);  // get by name may failed due to the cache cleanup
     assert(pTableMeta != NULL);
+
     pFinalInfo = tscAddTableMetaInfo(pNewQueryInfo, name, pTableMeta, pTableMetaInfo->vgroupList, pTableMetaInfo->tagColList);
   } else {  // transfer the ownership of pTableMeta to the newly create sql object.
     STableMetaInfo* pPrevInfo = tscGetTableMetaInfoFromCmd(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex, 0);