diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 9a029b0fea..0630704d93 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -45,7 +45,7 @@ extern "C" { #define ERROR_MSG_BUF_DEFAULT_SIZE 512 #define HEARTBEAT_INTERVAL 1500 // ms -#define SYNC_ON_TOP_OF_ASYNC 1 +#define SYNC_ON_TOP_OF_ASYNC 0 enum { RES_TYPE__QUERY = 1, @@ -266,8 +266,8 @@ extern SAppInfo appInfo; extern int32_t clientReqRefPool; extern int32_t clientConnRefPool; -extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code); -int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code); +__async_send_cb_fn_t getMsgRspHandle(int32_t msgType); + SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj); void* createTscObj(const char* user, const char* auth, const char* db, int32_t connType, SAppInstInfo* pAppInfo); @@ -293,8 +293,6 @@ void* openTransporter(const char* user, const char* auth, int32_t numOfThreads); bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType); void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet); -void initMsgHandleFp(); - TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db, uint16_t port, int connType); diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index c210a7ed39..75e1884360 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -278,7 +278,6 @@ void taos_init_imp(void) { return; } - initMsgHandleFp(); initQueryModuleMsgHandle(); rpcInit(); diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 85f6e85190..c71c7d3487 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -535,19 +535,20 @@ int32_t handleAlterTbExecRes(void* res, SCatalog* pCatalog) { return catalogUpdateTableMeta(pCatalog, (STableMetaRsp*)res); } -int32_t handleExecRes(SRequestObj* pRequest) { +int32_t handleQueryExecRsp(SRequestObj* pRequest) { if (NULL == pRequest->body.resInfo.execRes.res) { return TSDB_CODE_SUCCESS; } - int32_t code = 0; SCatalog* pCatalog = NULL; - code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); + SAppInstInfo* pAppInfo = getAppInfo(pRequest); + + int32_t code = catalogGetHandle(pAppInfo->clusterId, &pCatalog); if (code) { return code; } - SEpSet epset = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp); + SEpSet epset = getEpSet_s(&pAppInfo->mgmtEp); SQueryExecRes* pRes = &pRequest->body.resInfo.execRes; switch (pRes->msgType) { @@ -565,8 +566,9 @@ int32_t handleExecRes(SRequestObj* pRequest) { break; } default: - tscError("invalid exec result for request type %d", pRequest->type); - return TSDB_CODE_APP_ERROR; + tscError("0x%"PRIx64", invalid exec result for request type %d, reqId:0x%"PRIx64, pRequest->self, + pRequest->type, pRequest->requestId); + code = TSDB_CODE_APP_ERROR; } return code; @@ -578,13 +580,18 @@ void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) { STscObj* pTscObj = pRequest->pTscObj; if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code)) { - // todo do nothing in clear value in request tscDebug("0x%"PRIx64" client retry to handle the error, code:%s, reqId:0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId); pRequest->prevCode = code; doAsyncQuery(pRequest, true); return; } + if (code == TSDB_CODE_SUCCESS) { + code = handleQueryExecRsp(pRequest); + ASSERT(pRequest->code == TSDB_CODE_SUCCESS); + pRequest->code = code; + } + if (NEED_CLIENT_RM_TBLMETA_REQ(pRequest->type)) { removeMeta(pTscObj, pRequest->tableList); } @@ -623,7 +630,7 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue qDestroyQuery(pQuery); } - handleExecRes(pRequest); + handleQueryExecRsp(pRequest); if (NULL != pRequest && TSDB_CODE_SUCCESS != code) { pRequest->code = terrno; @@ -688,9 +695,6 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery) { if (TSDB_CODE_SUCCESS == code) { schedulerAsyncExecJob(pAppInfo->pTransporter, pNodeList, pRequest->body.pDag, &pRequest->body.queryJob, pRequest->sqlstr, pRequest->metric.start, schedulerExecCb, pRequest); - // if (NULL != pRes) { - // code = validateSversion(pRequest, pRes); - // } } //todo not to be released here @@ -884,7 +888,7 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) { pMsgSendInfo->requestObjRefId = pRequest->self; pMsgSendInfo->requestId = pRequest->requestId; - pMsgSendInfo->fp = handleRequestRspFp[TMSG_INDEX(pMsgSendInfo->msgType)]; + pMsgSendInfo->fp = getMsgRspHandle(pMsgSendInfo->msgType); pMsgSendInfo->param = pRequest; SConnectReq connectReq = {0}; diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index a15f9f17f7..d192b65bf0 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -634,7 +634,7 @@ void retrieveMetaCallback(SMetaData* pResultMeta, void* param, int32_t code) { // return to app directly taosMemoryFree(pWrapper); - tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:%" PRIx64, pRequest->self, tstrerror(code), + tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self, tstrerror(code), pRequest->requestId); pRequest->code = code; pRequest->body.queryFp(pRequest->body.param, pRequest, code); @@ -748,7 +748,6 @@ void doAsyncQuery(SRequestObj* pRequest, bool updateMetaForce) { pRequest->body.queryFp(pRequest->body.param, pRequest, code); } - static void fetchCallback(void* pResult, void* param, int32_t code) { SRequestObj* pRequest = (SRequestObj*) param; diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index cdac217084..14e2798d3b 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -21,8 +21,6 @@ #include "tdef.h" #include "tname.h" -int32_t (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code); - static void setErrno(SRequestObj* pRequest, int32_t code) { pRequest->code = code; terrno = code; @@ -107,10 +105,7 @@ SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pRequest) { assert(pRequest != NULL); pMsgSendInfo->msgInfo = pRequest->body.requestMsg; - - pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL) - ? genericRspCallback - : handleRequestRspFp[TMSG_INDEX(pRequest->type)]; + pMsgSendInfo->fp = getMsgRspHandle(pRequest->type); return pMsgSendInfo; } @@ -209,7 +204,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) { return 0; } -int32_t processCreateTableRsp(void* param, const SDataBuf* pMsg, int32_t code) { +int32_t processCreateSTableRsp(void* param, const SDataBuf* pMsg, int32_t code) { assert(pMsg != NULL && param != NULL); SRequestObj* pRequest = param; @@ -285,13 +280,21 @@ int32_t processAlterStbRsp(void* param, const SDataBuf* pMsg, int32_t code) { return code; } - -// todo refactor: this arraylist is too large -void initMsgHandleFp() { - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CONNECT)] = processConnectRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_DB)] = processCreateDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_USE_DB)] = processUseDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = processCreateTableRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = processDropDbRsp; - handleRequestRspFp[TMSG_INDEX(TDMT_MND_ALTER_STB)] = processAlterStbRsp; +__async_send_cb_fn_t getMsgRspHandle(int32_t msgType) { + switch (msgType) { + case TDMT_MND_CONNECT: + return processConnectRsp; + case TDMT_MND_CREATE_DB: + return processCreateDbRsp; + case TDMT_MND_USE_DB: + return processUseDbRsp; + case TDMT_MND_CREATE_STB: + return processCreateSTableRsp; + case TDMT_MND_DROP_DB: + return processDropDbRsp; + case TDMT_MND_ALTER_STB: + return processAlterStbRsp; + default: + return genericRspCallback; + } } diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index 1ce1a2590a..245089ddb7 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -1497,20 +1497,26 @@ char *buildRetension(SArray *pRetension) { int64_t v1 = getValOfDiffPrecision(p->freqUnit, p->freq); int64_t v2 = getValOfDiffPrecision(p->keepUnit, p->keep); - len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c,", v1, p->freqUnit, v2, p->keepUnit); - - p = taosArrayGet(pRetension, 1); - - v1 = getValOfDiffPrecision(p->freqUnit, p->freq); - v2 = getValOfDiffPrecision(p->keepUnit, p->keep); - len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c,", v1, p->freqUnit, v2, p->keepUnit); - - p = taosArrayGet(pRetension, 2); - - v1 = getValOfDiffPrecision(p->freqUnit, p->freq); - v2 = getValOfDiffPrecision(p->keepUnit, p->keep); len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit); + if (size > 1) { + len += sprintf(p1 + len, ","); + p = taosArrayGet(pRetension, 1); + + v1 = getValOfDiffPrecision(p->freqUnit, p->freq); + v2 = getValOfDiffPrecision(p->keepUnit, p->keep); + len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit); + } + + if (size > 2) { + len += sprintf(p1 + len, ","); + p = taosArrayGet(pRetension, 2); + + v1 = getValOfDiffPrecision(p->freqUnit, p->freq); + v2 = getValOfDiffPrecision(p->keepUnit, p->keep); + len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit); + } + varDataSetLen(p1, len); return p1; } diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index f1892d0040..6334d079c8 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2903,7 +2903,6 @@ static int32_t buildRollupAst(STranslateContext* pCxt, SCreateTableStmt* pStmt, } } - taosArrayDestroy(dbCfg.pRetensions); return code; }