diff --git a/src/client/inc/tscUtil.h b/src/client/inc/tscUtil.h index d6562f008de5ce8a42969633d53936d19fcc07ef..f954667287c02c951c71e7fd3d11162209fa9890 100644 --- a/src/client/inc/tscUtil.h +++ b/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); diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 52e2b97e1cd06fdc828336491f444080321055d4..a0eb1bd63c7767b774e58301625611e58817ec55 100644 --- a/src/client/inc/tsclient.h +++ b/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); diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index b03972bbfa2fc8f98251529c79bdbb88357b6cc1..e65acc2483c641de4a4318e8f792ba78b1d3f177 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -1048,7 +1048,7 @@ int tsParseInsertSql(SSqlObj *pSql) { str = pCmd->curSql; } - tscTrace("%p create data block list for submit data:%p, curSql:%p, pTableList:%p", pSql, pSql->cmd.pDataBlocks, pCmd->curSql, pCmd->pTableList); + tscTrace("%p create data block list for submit data:%p, pTableList:%p", pSql, pCmd->pDataBlocks, pCmd->pTableList); while (1) { int32_t index = 0; @@ -1088,22 +1088,16 @@ int tsParseInsertSql(SSqlObj *pSql) { goto _error; } - ptrdiff_t pos = pCmd->curSql - pSql->sqlstr; - if ((code = tscCheckIfCreateTable(&str, pSql)) != TSDB_CODE_SUCCESS) { /* - * For async insert, after get the table meta from server, the sql string will not be - * parsed using the new table meta to avoid the overhead cause by get table meta data information. - * And during the getMeterMetaCallback function, the sql string will be parsed from the - * interrupted position. + * After retrieving the table meta from server, the sql string will be parsed from the paused position. + * And during the getTableMetaCallback function, the sql string will be parsed from the paused position. */ if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) { - tscTrace("%p waiting for get table meta during insert, then resume from offset: %" PRId64 ", %s", pSql, - pos, pCmd->curSql); return code; } - tscError("%p async insert parse error, code:%d, reason:%s", pSql, code, tstrerror(code)); + tscError("%p async insert parse error, code:%s", pSql, tstrerror(code)); pCmd->curSql = NULL; goto _error; } @@ -1317,11 +1311,11 @@ int tsInsertInitialCheck(SSqlObj *pSql) { return TSDB_CODE_SUCCESS; } -int tsParseSql(SSqlObj *pSql, bool initialParse) { +int tsParseSql(SSqlObj *pSql, bool initial) { int32_t ret = TSDB_CODE_SUCCESS; SSqlCmd* pCmd = &pSql->cmd; - if (!pCmd->parseFinished) { + if ((!pCmd->parseFinished) && (!initial)) { tscTrace("%p resume to parse sql: %s", pSql, pCmd->curSql); } @@ -1330,12 +1324,12 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) { * Set the fp before parse the sql string, in case of getTableMeta failed, in which * the error handle callback function can rightfully restore the user-defined callback function (fp). */ - if (initialParse && (pSql->cmd.insertType != TSDB_QUERY_TYPE_STMT_INSERT)) { + if (initial && (pSql->cmd.insertType != TSDB_QUERY_TYPE_STMT_INSERT)) { pSql->fetchFp = pSql->fp; pSql->fp = (void(*)())tscHandleMultivnodeInsert; } - if (initialParse && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) { + if (initial && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) { return ret; } diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 87af5fb5f7363eb741ccc9d9ed2f0dffe5492c5e..f241e317cdc83b432a1a1d964dff80ea84b663bb 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -5755,6 +5755,7 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) { const char* msg7 = "illegal number of tables in from clause"; const char* msg8 = "too many columns in selection clause"; const char* msg9 = "TWA query requires both the start and end time"; + const char* msg10= "too many tables in from clause"; int32_t code = TSDB_CODE_SUCCESS; @@ -5790,6 +5791,10 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) { pQueryInfo->command = TSDB_SQL_SELECT; + if (pQuerySql->from->nExpr > 2) { + return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg10); + } + // set all query tables, which are maybe more than one. for (int32_t i = 0; i < pQuerySql->from->nExpr; ++i) { tVariant* pTableItem = &pQuerySql->from->a[i].pVar; diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 6e3602b69d5bd455255b2ebace20b5a797ab147a..b2e0c0107ea89a800550839055bb0b338fbf8160 100644 --- a/src/client/src/tscServer.c +++ b/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); } /* diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 47a602a6fb15c6461054cb034ad378b9ec6c891a..51b49304a8e72adb980ccbcf9368f68051f02745 100644 --- a/src/client/src/tscSql.c +++ b/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); } diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index 6c6284b4a6dca4e3fa7fc4337e6ea5fa0089a09c..87c75bd7e684ae942a4d7b49dd33efedf0c0d388 100644 --- a/src/client/src/tscSubquery.c +++ b/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; } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 4bb1fb3ed598b4131c1991de35a1b7b627db0418..ffe3dca9eb42b07de9c25d2128b999c6931fadcc 100644 --- a/src/client/src/tscUtil.c +++ b/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,46 @@ 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; + + pNew->sqlstr = strdup(pSql->sqlstr); + if (pNew->sqlstr == NULL) { + tscError("%p new subquery failed", pSql); + + free(pNew); + return NULL; + } + + 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)); @@ -1795,6 +1834,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);