From f12d252116ba2a2d2a23993fc40d9bca6497ebce Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 29 Jul 2021 18:08:02 +0800 Subject: [PATCH] [td-5175] --- src/client/src/tscAsync.c | 18 +++--- src/client/src/tscServer.c | 11 ++-- src/client/src/tscSubquery.c | 8 ++- src/client/src/tscUtil.c | 104 +++++++++++++++++++++++++++++++---- 4 files changed, 113 insertions(+), 28 deletions(-) diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index c8c9fe85e3..64f259ecaa 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -211,27 +211,27 @@ void taos_fetch_rows_a(TAOS_RES *tres, __async_cb_func_t fp, void *param) { pSql->fp = tscAsyncFetchRowsProxy; pSql->param = param; - if (pRes->qId == 0) { - tscError("qhandle is invalid"); - pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE; - tscAsyncResultOnError(pSql); - return; - } - tscResetForNextRetrieve(pRes); // handle outer query based on the already retrieved nest query results. SQueryInfo* pQueryInfo = tscGetQueryInfo(pCmd); if (pQueryInfo->pUpstream != NULL && taosArrayGetSize(pQueryInfo->pUpstream) > 0) { SSchedMsg schedMsg = {0}; - schedMsg.fp = doRetrieveSubqueryData; + schedMsg.fp = doRetrieveSubqueryData; schedMsg.ahandle = (void *)pSql; schedMsg.thandle = (void *)1; - schedMsg.msg = 0; + schedMsg.msg = 0; taosScheduleTask(tscQhandle, &schedMsg); return; } + if (pRes->qId == 0) { + tscError("qhandle is invalid"); + pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE; + tscAsyncResultOnError(pSql); + return; + } + if (pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE) { tscFetchDatablockForSubquery(pSql); } else if (pRes->completed) { diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index eaf397529b..a069e7639c 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -392,14 +392,17 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { // single table query error need to be handled here. if ((cmd == TSDB_SQL_SELECT || cmd == TSDB_SQL_UPDATE_TAGS_VAL) && - (((rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID || // change the retry procedure + (((rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID || rpcMsg->code == TSDB_CODE_VND_INVALID_VGROUP_ID)) || - rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || // change the retry procedure + rpcMsg->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || rpcMsg->code == TSDB_CODE_APP_NOT_READY)) { - if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, (TSDB_QUERY_TYPE_STABLE_SUBQUERY | TSDB_QUERY_TYPE_SUBQUERY | + // 1. super table subquery + // 2. nest queries are all not updated the tablemeta and retry parse the sql after cleanup local tablemeta/vgroup id buffer + if ((TSDB_QUERY_HAS_TYPE(pQueryInfo->type, (TSDB_QUERY_TYPE_STABLE_SUBQUERY | TSDB_QUERY_TYPE_SUBQUERY | TSDB_QUERY_TYPE_TAG_FILTER_QUERY)) && - !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_PROJECTION_QUERY)) { + !TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_PROJECTION_QUERY)) || + (pQueryInfo->pUpstream != NULL)) { // do nothing in case of super table subquery } else { pSql->retry += 1; diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index af7a18ca7a..8c20aed350 100644 --- a/src/client/src/tscSubquery.c +++ b/src/client/src/tscSubquery.c @@ -2705,7 +2705,6 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO // release allocated resource tscDestroyGlobalMergerEnv(trsupport->pExtMemBuffer, trsupport->pOrderDescriptor, pState->numOfSub); - tscFreeRetrieveSup(pSql); // in case of second stage join subquery, invoke its callback function instead of regular QueueAsyncRes @@ -2716,10 +2715,13 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO int32_t code = pParentSql->res.code; if ((code == TSDB_CODE_TDB_INVALID_TABLE_ID || code == TSDB_CODE_VND_INVALID_VGROUP_ID) && pParentSql->retry < pParentSql->maxRetry) { // remove the cached tableMeta and vgroup id list, and then parse the sql again - STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pParentSql->cmd, 0); + SSqlCmd* pParentCmd = &pParentSql->cmd; + STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pParentCmd, 0); tscRemoveTableMetaBuf(pTableMetaInfo, pParentSql->self); - tscResetSqlCmd(&pParentSql->cmd, true); + pParentCmd->pTableMetaMap = tscCleanupTableMetaMap(pParentCmd->pTableMetaMap); + pParentCmd->pTableMetaMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); + pParentSql->res.code = TSDB_CODE_SUCCESS; pParentSql->retry++; diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index b9a9333fd7..d4425f7ad2 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -3829,13 +3829,64 @@ static void tscSubqueryRetrieveCallback(void* param, TAOS_RES* tres, int code) { } } -// todo handle the failure static void tscSubqueryCompleteCallback(void* param, TAOS_RES* tres, int code) { + SSqlObj* pSql = tres; + SRetrieveSupport* ps = param; + + if (pSql->res.code != TSDB_CODE_SUCCESS) { + SSqlObj* pParentSql = ps->pParentSql; + + int32_t index = ps->subqueryIndex; + bool ret = subAndCheckDone(pSql, pParentSql, index); + + tfree(ps); + pSql->param = NULL; + + if (!ret) { + tscDebug("0x%"PRIx64" sub:0x%"PRIx64" orderOfSub:%d completed, not all subquery finished", pParentSql->self, pSql->self, index); + return; + } + + // todo refactor + tscDebug("0x%"PRIx64" all subquery response received, retry", pParentSql->self); + + SSqlCmd* pParentCmd = &pParentSql->cmd; + STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pParentCmd, 0); + tscRemoveTableMetaBuf(pTableMetaInfo, pParentSql->self); + + pParentCmd->pTableMetaMap = tscCleanupTableMetaMap(pParentCmd->pTableMetaMap); + pParentCmd->pTableMetaMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); + + pParentSql->res.code = TSDB_CODE_SUCCESS; + pParentSql->retry++; + + tscDebug("0x%"PRIx64" retry parse sql and send query, prev error: %s, retry:%d", pParentSql->self, + tstrerror(code), pParentSql->retry); + + code = tsParseSql(pParentSql, true); + if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { + return; + } + + if (code != TSDB_CODE_SUCCESS) { + pParentSql->res.code = code; + tscAsyncResultOnError(pParentSql); + return; + } + + SQueryInfo *pQueryInfo = tscGetQueryInfo(pParentCmd); + executeQuery(pParentSql, pQueryInfo); + return; + } + taos_fetch_rows_a(tres, tscSubqueryRetrieveCallback, param); } // do execute the query according to the query execution plan void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) { + int32_t code = TSDB_CODE_SUCCESS; + int32_t numOfInit = 0; + if (pSql->cmd.command == TSDB_SQL_RETRIEVE_EMPTY_RESULT) { (*pSql->fp)(pSql->param, pSql, 0); return; @@ -3850,7 +3901,12 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) { pSql->pSubs = calloc(pSql->subState.numOfSub, POINTER_BYTES); pSql->subState.states = calloc(pSql->subState.numOfSub, sizeof(int8_t)); - pthread_mutex_init(&pSql->subState.mutex, NULL); + code = pthread_mutex_init(&pSql->subState.mutex, NULL); + + if (pSql->pSubs == NULL || pSql->subState.states == NULL || code != TSDB_CODE_SUCCESS) { + code = TSDB_CODE_TSC_OUT_OF_MEMORY; + goto _error; + } for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) { SQueryInfo* pSub = taosArrayGetP(pQueryInfo->pUpstream, i); @@ -3858,45 +3914,69 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) { pSql->cmd.active = pSub; pSql->cmd.command = TSDB_SQL_SELECT; - // TODO handle memory failure SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj)); if (pNew == NULL) { - terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; - // return NULL; + code = TSDB_CODE_TSC_OUT_OF_MEMORY; + goto _error; } - pNew->pTscObj = pSql->pTscObj; + pNew->pTscObj = pSql->pTscObj; pNew->signature = pNew; - pNew->sqlstr = strdup(pSql->sqlstr); // todo refactor - pNew->fp = tscSubqueryCompleteCallback; + pNew->sqlstr = strdup(pSql->sqlstr); + pNew->fp = tscSubqueryCompleteCallback; + pNew->maxRetry = pSql->maxRetry; tsem_init(&pNew->rspSem, 0, 0); SRetrieveSupport* ps = calloc(1, sizeof(SRetrieveSupport)); // todo use object id + if (ps == NULL) { + tscFreeSqlObj(pNew); + goto _error; + } + ps->pParentSql = pSql; ps->subqueryIndex = i; pNew->param = ps; pSql->pSubs[i] = pNew; - registerSqlObj(pNew); SSqlCmd* pCmd = &pNew->cmd; pCmd->command = TSDB_SQL_SELECT; - if (tscAddQueryInfo(pCmd) != TSDB_CODE_SUCCESS) { + if ((code = tscAddQueryInfo(pCmd)) != TSDB_CODE_SUCCESS) { + goto _error; } SQueryInfo* pNewQueryInfo = tscGetQueryInfo(pCmd); tscQueryInfoCopy(pNewQueryInfo, pSub); + numOfInit++; + } + + for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) { + SSqlObj* psub = pSql->pSubs[i]; + registerSqlObj(psub); // create sub query to handle the sub query. - executeQuery(pNew, pNewQueryInfo); + SQueryInfo* pq = tscGetQueryInfo(&psub->cmd); + executeQuery(psub, pq); } - // merge sub query result and generate final results return; } pSql->cmd.active = pQueryInfo; doExecuteQuery(pSql, pQueryInfo); + return; + + _error: + for(int32_t i = 0; i < numOfInit; ++i) { + SSqlObj* p = pSql->pSubs[i]; + tscFreeSqlObj(p); + } + + pSql->res.code = code; + pSql->subState.numOfSub = 0; // not initialized sub query object will not be freed + tfree(pSql->subState.states); + tfree(pSql->pSubs); + tscAsyncResultOnError(pSql); } int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid) { -- GitLab