From c7d115d8aaa6f20b26575ec73ccdb269f4ec875b Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 4 Mar 2024 14:58:13 +0800 Subject: [PATCH 1/5] fix:add excluded msg for delete in tmq --- include/common/tmsg.h | 3 +++ source/client/src/clientRawBlockWrite.c | 2 ++ source/common/src/tmsg.c | 9 +++++++++ source/dnode/vnode/src/tq/tqUtil.c | 3 +-- source/libs/qworker/src/qwMsg.c | 1 + source/libs/scheduler/src/schRemote.c | 1 + tests/script/sh/deploy.sh | 2 +- utils/test/c/tmq_taosx_ci.c | 1 + 8 files changed, 19 insertions(+), 3 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 587e2f9f3e..e400698f69 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -3813,8 +3813,10 @@ typedef struct { uint32_t phyLen; char* sql; char* msg; + int8_t source; } SVDeleteReq; +extern int8_t deleteFromTaosx; int32_t tSerializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); int32_t tDeserializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); @@ -3834,6 +3836,7 @@ typedef struct SDeleteRes { char tableFName[TSDB_TABLE_NAME_LEN]; char tsColName[TSDB_COL_NAME_LEN]; int64_t ctimeMs; // fill by vnode + int8_t source; } SDeleteRes; int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes); diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index f143624bab..3857e47f57 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -1256,7 +1256,9 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) { snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName, req.tsColName, req.skey, req.tsColName, req.ekey); + deleteFromTaosx = TD_REQ_FROM_TAOX; TAOS_RES* res = taos_query(taos, sql); + deleteFromTaosx = TD_REQ_FROM_APP; SRequestObj* pRequest = (SRequestObj*)res; code = pRequest->code; if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_PAR_GET_META_ERROR) { diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 85f5d462c7..e7414c6004 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -7148,6 +7148,7 @@ int32_t tDecodeSVDropTSmaReq(SDecoder *pCoder, SVDropTSmaReq *pReq) { return 0; } +int8_t deleteFromTaosx = TD_REQ_FROM_APP; int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) { int32_t headLen = sizeof(SMsgHead); if (buf != NULL) { @@ -7165,6 +7166,7 @@ int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) { if (tEncodeU32(&encoder, pReq->sqlLen) < 0) return -1; if (tEncodeCStr(&encoder, pReq->sql) < 0) return -1; if (tEncodeBinary(&encoder, pReq->msg, pReq->phyLen) < 0) return -1; + if (tEncodeI8(&encoder, pReq->source) < 0) return -1; tEndEncode(&encoder); int32_t tlen = encoder.pos; @@ -7201,6 +7203,9 @@ int32_t tDeserializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) { if (tDecodeBinaryAlloc(&decoder, (void **)&pReq->msg, &msgLen) < 0) return -1; pReq->phyLen = msgLen; + if (!tDecodeIsEnd(&decoder)) { + if (tDecodeI8(&decoder, &pReq->source) < 0) return -1; + } tEndDecode(&decoder); tDecoderClear(&decoder); @@ -8400,6 +8405,7 @@ int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) { if (tEncodeCStr(pCoder, pRes->tableFName) < 0) return -1; if (tEncodeCStr(pCoder, pRes->tsColName) < 0) return -1; if (tEncodeI64(pCoder, pRes->ctimeMs) < 0) return -1; + if (tEncodeI8(pCoder, pRes->source) < 0) return -1; return 0; } @@ -8424,6 +8430,9 @@ int32_t tDecodeDeleteRes(SDecoder *pCoder, SDeleteRes *pRes) { if (!tDecodeIsEnd(pCoder)) { if (tDecodeI64(pCoder, &pRes->ctimeMs) < 0) return -1; } + if (!tDecodeIsEnd(pCoder)) { + if (tDecodeI8(pCoder, &pRes->source) < 0) return -1; + } return 0; } diff --git a/source/dnode/vnode/src/tq/tqUtil.c b/source/dnode/vnode/src/tq/tqUtil.c index dad1211326..6029575e2c 100644 --- a/source/dnode/vnode/src/tq/tqUtil.c +++ b/source/dnode/vnode/src/tq/tqUtil.c @@ -263,8 +263,7 @@ static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle, } else if (pHead->msgType == TDMT_VND_CREATE_STB || pHead->msgType == TDMT_VND_ALTER_STB) { PROCESS_EXCLUDED_MSG(SVCreateStbReq, tDecodeSVCreateStbReq) } else if (pHead->msgType == TDMT_VND_DELETE) { - fetchVer++; - continue; + PROCESS_EXCLUDED_MSG(SDeleteRes, tDecodeDeleteRes) } } diff --git a/source/libs/qworker/src/qwMsg.c b/source/libs/qworker/src/qwMsg.c index 66ec460861..faa90dcbf8 100644 --- a/source/libs/qworker/src/qwMsg.c +++ b/source/libs/qworker/src/qwMsg.c @@ -715,6 +715,7 @@ int32_t qWorkerProcessDeleteMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SD uint64_t tId = req.taskId; int64_t rId = 0; int32_t eId = -1; + pRes->source = req.source; SQWMsg qwMsg = {.node = node, .msg = req.msg, .msgLen = req.phyLen, .connInfo = pMsg->info}; QW_SCH_TASK_DLOG("processDelete start, node:%p, handle:%p, sql:%s", node, pMsg->info.handle, req.sql); diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index 1c0b31109e..f6465316d2 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -1086,6 +1086,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, req.sqlLen = strlen(pJob->sql); req.sql = (char *)pJob->sql; req.msg = pTask->msg; + req.source = deleteFromTaosx; msgSize = tSerializeSVDeleteReq(NULL, 0, &req); msg = taosMemoryCalloc(1, msgSize); if (NULL == msg) { diff --git a/tests/script/sh/deploy.sh b/tests/script/sh/deploy.sh index 3b3d275a07..9af2525c74 100755 --- a/tests/script/sh/deploy.sh +++ b/tests/script/sh/deploy.sh @@ -117,7 +117,7 @@ echo "supportVnodes 1024" >> $TAOS_CFG echo "statusInterval 1" >> $TAOS_CFG echo "dataDir $DATA_DIR" >> $TAOS_CFG echo "logDir $LOG_DIR" >> $TAOS_CFG -echo "debugFlag 0" >> $TAOS_CFG +echo "debugFlag 135" >> $TAOS_CFG echo "tmrDebugFlag 131" >> $TAOS_CFG echo "uDebugFlag 143" >> $TAOS_CFG echo "rpcDebugFlag 143" >> $TAOS_CFG diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index 2257089f06..5012e50bab 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -574,6 +574,7 @@ tmq_t* build_consumer() { tmq_conf_set(conf, "msg.with.table.name", "true"); tmq_conf_set(conf, "enable.auto.commit", "true"); tmq_conf_set(conf, "auto.offset.reset", "earliest"); + tmq_conf_set(conf, "msg.consume.excluded", "1"); if (g_conf.snapShot) { tmq_conf_set(conf, "experimental.snapshot.enable", "true"); From af6d6b69c3380385a6effe88d0c557cc4735e693 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Tue, 5 Mar 2024 11:25:43 +0800 Subject: [PATCH 2/5] fix:add excluded msg for delete in tmq --- include/common/tmsg.h | 1 - include/libs/scheduler/scheduler.h | 1 + source/client/inc/clientInt.h | 5 +++-- source/client/src/clientImpl.c | 9 ++++++--- source/client/src/clientMain.c | 6 +++--- source/client/src/clientRawBlockWrite.c | 4 +--- source/common/src/tmsg.c | 1 - source/libs/scheduler/inc/schInt.h | 1 + source/libs/scheduler/src/schJob.c | 1 + source/libs/scheduler/src/schRemote.c | 2 +- 10 files changed, 17 insertions(+), 14 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index e400698f69..4fa244a07c 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -3816,7 +3816,6 @@ typedef struct { int8_t source; } SVDeleteReq; -extern int8_t deleteFromTaosx; int32_t tSerializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); int32_t tDeserializeSVDeleteReq(void* buf, int32_t bufLen, SVDeleteReq* pReq); diff --git a/include/libs/scheduler/scheduler.h b/include/libs/scheduler/scheduler.h index 958d63349d..952af3c443 100644 --- a/include/libs/scheduler/scheduler.h +++ b/include/libs/scheduler/scheduler.h @@ -78,6 +78,7 @@ typedef struct SSchedulerReq { void* chkKillParam; SExecResult* pExecRes; void** pFetchRes; + int8_t source; } SSchedulerReq; int32_t schedulerInit(void); diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index f73d121b15..6cc2a62236 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -284,6 +284,7 @@ typedef struct SRequestObj { void* pWrapper; SMetaData parseMeta; char* effectiveUser; + int8_t source; } SRequestObj; typedef struct SSyncQueryParam { @@ -306,10 +307,10 @@ void doFreeReqResultInfo(SReqResultInfo* pResInfo); int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq); void syncCatalogFn(SMetaData* pResult, void* param, int32_t code); -TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly); +TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly, int8_t source); TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly, int64_t reqid); -void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly); +void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int8_t source); void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int64_t reqid); void taosAsyncFetchImpl(SRequestObj *pRequest, __taos_async_fn_t fp, void *param); diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 971ed407f9..aa18ef1c94 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -729,6 +729,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList .chkKillFp = chkRequestKilled, .chkKillParam = (void*)pRequest->self, .pExecRes = &res, + .source = pRequest->source, }; int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob); @@ -1198,6 +1199,7 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat .chkKillFp = chkRequestKilled, .chkKillParam = (void*)pRequest->self, .pExecRes = NULL, + .source = pRequest->source, }; code = schedulerExecJob(&req, &pRequest->body.queryJob); taosArrayDestroy(pNodeList); @@ -2461,7 +2463,7 @@ void syncQueryFn(void* param, void* res, int32_t code) { tsem_post(&pParam->sem); } -void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly) { +void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int8_t source) { if (sql == NULL || NULL == fp) { terrno = TSDB_CODE_INVALID_PARA; if (fp) { @@ -2487,6 +2489,7 @@ void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, return; } + pRequest->source = source; pRequest->body.queryFp = fp; doAsyncQuery(pRequest, false); } @@ -2521,7 +2524,7 @@ void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_ doAsyncQuery(pRequest, false); } -TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) { +TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly, int8_t source) { if (NULL == taos) { terrno = TSDB_CODE_TSC_DISCONNECTED; return NULL; @@ -2536,7 +2539,7 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) { } tsem_init(¶m->sem, 0, 0); - taosAsyncQueryImpl(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly); + taosAsyncQueryImpl(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, source); tsem_wait(¶m->sem); SRequestObj* pRequest = NULL; diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index f6aef5aa26..eacad0f18f 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -401,7 +401,7 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) { return pResInfo->userFields; } -TAOS_RES *taos_query(TAOS *taos, const char *sql) { return taosQueryImpl(taos, sql, false); } +TAOS_RES *taos_query(TAOS *taos, const char *sql) { return taosQueryImpl(taos, sql, false, TD_REQ_FROM_APP); } TAOS_RES *taos_query_with_reqid(TAOS *taos, const char *sql, int64_t reqid) { return taosQueryImplWithReqid(taos, sql, false, reqid); } @@ -827,7 +827,7 @@ int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) { } int taos_validate_sql(TAOS *taos, const char *sql) { - TAOS_RES *pObj = taosQueryImpl(taos, sql, true); + TAOS_RES *pObj = taosQueryImpl(taos, sql, true, TD_REQ_FROM_APP); int code = taos_errno(pObj); @@ -1125,7 +1125,7 @@ void continueInsertFromCsv(SSqlCallbackWrapper *pWrapper, SRequestObj *pRequest) void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) { int64_t connId = *(int64_t *)taos; tscDebug("taos_query_a start with sql:%s", sql); - taosAsyncQueryImpl(connId, sql, fp, param, false); + taosAsyncQueryImpl(connId, sql, fp, param, false, TD_REQ_FROM_APP); tscDebug("taos_query_a end with sql:%s", sql); } diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index 3857e47f57..adc8c361cd 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -1256,9 +1256,7 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) { snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName, req.tsColName, req.skey, req.tsColName, req.ekey); - deleteFromTaosx = TD_REQ_FROM_TAOX; - TAOS_RES* res = taos_query(taos, sql); - deleteFromTaosx = TD_REQ_FROM_APP; + TAOS_RES* res = taosQueryImpl(taos, sql, false, TD_REQ_FROM_TAOX); SRequestObj* pRequest = (SRequestObj*)res; code = pRequest->code; if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_PAR_GET_META_ERROR) { diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index e7414c6004..467300352c 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -7148,7 +7148,6 @@ int32_t tDecodeSVDropTSmaReq(SDecoder *pCoder, SVDropTSmaReq *pReq) { return 0; } -int8_t deleteFromTaosx = TD_REQ_FROM_APP; int32_t tSerializeSVDeleteReq(void *buf, int32_t bufLen, SVDeleteReq *pReq) { int32_t headLen = sizeof(SMsgHead); if (buf != NULL) { diff --git a/source/libs/scheduler/inc/schInt.h b/source/libs/scheduler/inc/schInt.h index 1c90e61ea3..276b668717 100644 --- a/source/libs/scheduler/inc/schInt.h +++ b/source/libs/scheduler/inc/schInt.h @@ -304,6 +304,7 @@ typedef struct SSchJob { SSchResInfo userRes; char *sql; SQueryProfileSummary summary; + int8_t source; } SSchJob; typedef struct SSchTaskCtx { diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c index e50ec64d54..48aab63ba3 100644 --- a/source/libs/scheduler/src/schJob.c +++ b/source/libs/scheduler/src/schJob.c @@ -746,6 +746,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) { pJob->chkKillParam = pReq->chkKillParam; pJob->userRes.execFp = pReq->execFp; pJob->userRes.cbParam = pReq->cbParam; + pJob->source = pReq->source; if (pReq->pNodeList == NULL || taosArrayGetSize(pReq->pNodeList) <= 0) { qDebug("QID:0x%" PRIx64 " input exec nodeList is empty", pReq->pDag->queryId); diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index f6465316d2..39273ffa50 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -1086,7 +1086,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, req.sqlLen = strlen(pJob->sql); req.sql = (char *)pJob->sql; req.msg = pTask->msg; - req.source = deleteFromTaosx; + req.source = pJob->source; msgSize = tSerializeSVDeleteReq(NULL, 0, &req); msg = taosMemoryCalloc(1, msgSize); if (NULL == msg) { From cdfbbc298535858df1a8a65de45803cee3855b6e Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 5 Mar 2024 13:28:34 +0800 Subject: [PATCH 3/5] fix(stream): set the dest epset from mnode, and set the retry error code for trans. --- source/dnode/mnode/impl/inc/mndStream.h | 2 +- source/dnode/mnode/impl/src/mndStream.c | 2 +- source/dnode/mnode/impl/src/mndStreamUtil.c | 16 ++++++++++++---- source/libs/stream/src/streamTask.c | 2 +- 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/source/dnode/mnode/impl/inc/mndStream.h b/source/dnode/mnode/impl/inc/mndStream.h index 57fd187da3..5307ff4b05 100644 --- a/source/dnode/mnode/impl/inc/mndStream.h +++ b/source/dnode/mnode/impl/inc/mndStream.h @@ -111,7 +111,7 @@ STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, SRpcMsg *pReq, const int32_t mndPersistTransLog(SStreamObj *pStream, STrans *pTrans, int32_t status); SSdbRaw *mndStreamActionEncode(SStreamObj *pStream); void killAllCheckpointTrans(SMnode *pMnode, SVgroupChangeInfo *pChangeInfo); -int32_t mndStreamSetUpdateEpsetAction(SStreamObj *pStream, SVgroupChangeInfo *pInfo, STrans *pTrans); +int32_t mndStreamSetUpdateEpsetAction(SMnode *pMnode, SStreamObj *pStream, SVgroupChangeInfo *pInfo, STrans *pTrans); SStreamObj *mndGetStreamObj(SMnode *pMnode, int64_t streamId); int32_t extractNodeEpset(SMnode *pMnode, SEpSet *pEpSet, bool *hasEpset, int32_t taskId, int32_t nodeId); diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 3ef2f64df7..fca757006d 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -1811,7 +1811,7 @@ static int32_t mndProcessVgroupChange(SMnode *pMnode, SVgroupChangeInfo *pChange mDebug("stream:0x%" PRIx64 " %s involved node changed, create update trans, transId:%d", pStream->uid, pStream->name, pTrans->id); - int32_t code = mndStreamSetUpdateEpsetAction(pStream, pChangeInfo, pTrans); + int32_t code = mndStreamSetUpdateEpsetAction(pMnode, pStream, pChangeInfo, pTrans); // todo: not continue, drop all and retry again if (code != TSDB_CODE_SUCCESS) { diff --git a/source/dnode/mnode/impl/src/mndStreamUtil.c b/source/dnode/mnode/impl/src/mndStreamUtil.c index 1ae85a2cc6..2b8fcee9fd 100644 --- a/source/dnode/mnode/impl/src/mndStreamUtil.c +++ b/source/dnode/mnode/impl/src/mndStreamUtil.c @@ -462,14 +462,22 @@ static int32_t doBuildStreamTaskUpdateMsg(void **pBuf, int32_t *pLen, SVgroupCha return TSDB_CODE_SUCCESS; } -static int32_t doSetUpdateTaskAction(STrans *pTrans, SStreamTask *pTask, SVgroupChangeInfo *pInfo) { +static int32_t doSetUpdateTaskAction(SMnode *pMnode, STrans *pTrans, SStreamTask *pTask, SVgroupChangeInfo *pInfo) { void *pBuf = NULL; int32_t len = 0; streamTaskUpdateEpsetInfo(pTask, pInfo->pUpdateNodeList); doBuildStreamTaskUpdateMsg(&pBuf, &len, pInfo, pTask->info.nodeId, &pTask->id, pTrans->id); - int32_t code = setTransAction(pTrans, pBuf, len, TDMT_VND_STREAM_TASK_UPDATE, &pTask->info.epSet, 0); + SEpSet epset = {0}; + bool hasEpset = false; + int32_t code = extractNodeEpset(pMnode, &epset, &hasEpset, pTask->id.taskId, pTask->info.nodeId); + if (code != TSDB_CODE_SUCCESS || !hasEpset) { + terrno = code; + return code; + } + + code = setTransAction(pTrans, pBuf, len, TDMT_VND_STREAM_TASK_UPDATE, &epset, TSDB_CODE_VND_INVALID_VGROUP_ID); if (code != TSDB_CODE_SUCCESS) { taosMemoryFree(pBuf); } @@ -478,14 +486,14 @@ static int32_t doSetUpdateTaskAction(STrans *pTrans, SStreamTask *pTask, SVgroup } // build trans to update the epset -int32_t mndStreamSetUpdateEpsetAction(SStreamObj *pStream, SVgroupChangeInfo *pInfo, STrans *pTrans) { +int32_t mndStreamSetUpdateEpsetAction(SMnode *pMnode, SStreamObj *pStream, SVgroupChangeInfo *pInfo, STrans *pTrans) { mDebug("stream:0x%" PRIx64 " set tasks epset update action", pStream->uid); taosWLockLatch(&pStream->lock); SStreamTaskIter *pIter = createStreamTaskIter(pStream); while (streamTaskIterNextTask(pIter)) { SStreamTask *pTask = streamTaskIterGetCurrent(pIter); - int32_t code = doSetUpdateTaskAction(pTrans, pTask, pInfo); + int32_t code = doSetUpdateTaskAction(pMnode, pTrans, pTask, pInfo); if (code != TSDB_CODE_SUCCESS) { destroyStreamTaskIter(pIter); taosWUnLockLatch(&pStream->lock); diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 8be5b94096..6fd96f651e 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -39,7 +39,7 @@ static int32_t doUpdateTaskEpset(SStreamTask* pTask, int32_t nodeId, SEpSet* pEp stDebug("s-task:0x%x (vgId:%d) self node epset is updated %s", pTask->id.taskId, nodeId, buf); } - // check for the dispath info and the upstream task info + // check for the dispatch info and the upstream task info int32_t level = pTask->info.taskLevel; if (level == TASK_LEVEL__SOURCE) { streamTaskUpdateDownstreamInfo(pTask, nodeId, pEpSet); From 254afc070c8050cfd3fb5f8fddff4eff7841ffec Mon Sep 17 00:00:00 2001 From: Shungang Li Date: Tue, 5 Mar 2024 15:20:11 +0800 Subject: [PATCH 4/5] fix: dynmaic config supportvnodes in ent --- source/common/src/tglobal.c | 2 +- source/dnode/mnode/impl/src/mndDnode.c | 28 +++++--------------------- 2 files changed, 6 insertions(+), 24 deletions(-) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 55df80ca44..fde8313228 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -586,7 +586,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { tsNumOfSupportVnodes = tsNumOfCores * 2; tsNumOfSupportVnodes = TMAX(tsNumOfSupportVnodes, 2); - if (cfgAddInt32(pCfg, "supportVnodes", tsNumOfSupportVnodes, 0, 4096, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; + if (cfgAddInt32(pCfg, "supportVnodes", tsNumOfSupportVnodes, 0, 4096, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1; if (cfgAddInt32(pCfg, "statusInterval", tsStatusInterval, 1, 30, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1; if (cfgAddInt32(pCfg, "minSlidingTime", tsMinSlidingTime, 1, 1000000, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT) != 0) diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c index 79a5f5fd83..befb6d3521 100644 --- a/source/dnode/mnode/impl/src/mndDnode.c +++ b/source/dnode/mnode/impl/src/mndDnode.c @@ -610,7 +610,7 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { for(int32_t j = 0; j < tagSize; j++){ SJson* item = tjsonGetArrayItem(arrayTag, j); - *(labels + j) = taosMemoryMalloc(MONITOR_TAG_NAME_LEN); + *(labels + j) = taosMemoryMalloc(MONITOR_TAG_NAME_LEN); tjsonGetStringValue(item, "name", *(labels + j)); *(sample_labels + j) = taosMemoryMalloc(MONITOR_TAG_VALUE_LEN); @@ -626,7 +626,7 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { for(int32_t j = 0; j < metricLen; j++){ SJson *item = tjsonGetArrayItem(metrics, j); - char name[MONITOR_METRIC_NAME_LEN] = {0}; + char name[MONITOR_METRIC_NAME_LEN] = {0}; tjsonGetStringValue(item, "name", name); double value = 0; @@ -636,7 +636,7 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { tjsonGetDoubleValue(item, "type", &type); int32_t metricNameLen = strlen(name) + strlen(tableName) + 2; - char* metricName = taosMemoryMalloc(metricNameLen); + char* metricName = taosMemoryMalloc(metricNameLen); memset(metricName, 0, metricNameLen); sprintf(metricName, "%s:%s", tableName, name); @@ -669,7 +669,7 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { else{ mTrace("get metric from registry:%p", metric); } - + if(type == 0){ taos_counter_add(metric, value, (const char**)sample_labels); } @@ -689,7 +689,7 @@ static int32_t mndProcessStatisReq(SRpcMsg *pReq) { taosMemoryFreeClear(labels); } } - + } code = 0; @@ -1409,24 +1409,6 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) { if (strcasecmp(cfgReq.config, "resetlog") == 0) { strcpy(dcfgReq.config, "resetlog"); #ifdef TD_ENTERPRISE - } else if (strncasecmp(cfgReq.config, "supportvnodes", 13) == 0) { - int32_t optLen = strlen("supportvnodes"); - int32_t flag = -1; - int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag); - if (code < 0) return code; - - if (flag < 0 || flag > 4096) { - mError("dnode:%d, failed to config supportVnodes since value:%d. Valid range: [0, 4096]", cfgReq.dnodeId, flag); - terrno = TSDB_CODE_OUT_OF_RANGE; - goto _err_out; - } - if (flag == 0) { - flag = tsNumOfCores * 2; - } - flag = TMAX(flag, 2); - - strcpy(dcfgReq.config, "supportvnodes"); - snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag); } else if (strncasecmp(cfgReq.config, "s3blocksize", 11) == 0) { int32_t optLen = strlen("s3blocksize"); int32_t flag = -1; From 45e779d464cf41fb60aa7e51bf940d6255c0171f Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 5 Mar 2024 15:48:58 +0800 Subject: [PATCH 5/5] refactor: do some internal refactor. --- source/libs/stream/src/streamTask.c | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 6fd96f651e..4dc01ee9b4 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -622,6 +622,7 @@ void streamTaskSetFixedDownstreamInfo(SStreamTask* pTask, const SStreamTask* pDo void streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet) { char buf[512] = {0}; EPSET_TO_STR(pEpSet, buf); + int32_t id = pTask->id.taskId; int8_t type = pTask->outputInfo.type; if (type == TASK_OUTPUT__SHUFFLE_DISPATCH) { @@ -633,8 +634,8 @@ void streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SE if (pVgInfo->vgId == nodeId) { epsetAssign(&pVgInfo->epSet, pEpSet); - stDebug("s-task:0x%x update the dispatch info, task:0x%x(nodeId:%d) newEpset:%s", pTask->id.taskId, - pVgInfo->taskId, nodeId, buf); + stDebug("s-task:0x%x update the dispatch info, task:0x%x(nodeId:%d) newEpset:%s", id, pVgInfo->taskId, nodeId, + buf); break; } } @@ -642,8 +643,8 @@ void streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SE STaskDispatcherFixed* pDispatcher = &pTask->outputInfo.fixedDispatcher; if (pDispatcher->nodeId == nodeId) { epsetAssign(&pDispatcher->epSet, pEpSet); - stDebug("s-task:0x%x update the dispatch info, task:0x%x(nodeId:%d) newEpSet:%s", pTask->id.taskId, - pDispatcher->taskId, nodeId, buf); + stDebug("s-task:0x%x update the dispatch info, task:0x%x(nodeId:%d) newEpset:%s", id, pDispatcher->taskId, nodeId, + buf); } } else { // do nothing