From db1d75bb48bfee6eb1a3e0f2ecce1c3629e73cf7 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 27 Apr 2023 09:28:20 +0800 Subject: [PATCH] refactor: do some internal refactor. --- include/common/tmsg.h | 2 +- source/common/src/tdatablock.c | 2 +- source/common/src/tmsg.c | 2 +- source/dnode/vnode/inc/vnode.h | 11 +++----- source/dnode/vnode/src/sma/smaRollup.c | 4 +-- source/dnode/vnode/src/sma/smaTimeRange.c | 2 +- source/dnode/vnode/src/tq/tqRead.c | 34 +++++++++++------------ source/dnode/vnode/src/tq/tqScan.c | 4 +-- source/dnode/vnode/src/tq/tqSink.c | 6 ++-- source/dnode/vnode/src/vnd/vnodeSvr.c | 2 +- source/libs/executor/src/dataInserter.c | 6 ++-- source/libs/executor/src/scanoperator.c | 29 ++++++++++--------- source/libs/parser/src/parInsertUtil.c | 2 +- 13 files changed, 50 insertions(+), 56 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 02c097b8d0..8a5f3a72af 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -3423,7 +3423,7 @@ typedef struct { int32_t tEncodeSSubmitReq2(SEncoder* pCoder, const SSubmitReq2* pReq); int32_t tDecodeSSubmitReq2(SDecoder* pCoder, SSubmitReq2* pReq); void tDestroySSubmitTbData(SSubmitTbData* pTbData, int32_t flag); -void tDestroySSubmitReq2(SSubmitReq2* pReq, int32_t flag); +void tDestroySSubmitReq(SSubmitReq2* pReq, int32_t flag); typedef struct { int32_t affectedRows; diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index f4d2ed01b0..3558feaa66 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -2388,7 +2388,7 @@ _end: if (terrno != 0) { *ppReq = NULL; if (pReq) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFreeClear(pReq); } diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index d9802244b7..5f8236b7a9 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -7436,7 +7436,7 @@ void tDestroySSubmitTbData(SSubmitTbData *pTbData, int32_t flag) { } } -void tDestroySSubmitReq2(SSubmitReq2 *pReq, int32_t flag) { +void tDestroySSubmitReq(SSubmitReq2 *pReq, int32_t flag) { if (pReq->aSubmitTbData == NULL) return; int32_t nSubmitTbData = TARRAY_SIZE(pReq->aSubmitTbData); diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index aecfb9c3e5..3bcb1c9d2e 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -259,17 +259,14 @@ int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList); int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList); int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id); -void tqNextBlock(STqReader *pReader, SFetchRet *ret); +int32_t tqNextBlock(STqReader *pReader, SSDataBlock* pBlock); int32_t extractSubmitMsgFromWal(SWalReader *pReader, SPackedData *pPackedData); int32_t tqReaderSetSubmitMsg(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver); -// int32_t tqReaderSetDataMsg(STqReader *pReader, const SSubmitReq *pMsg, int64_t ver); -bool tqNextDataBlock(STqReader *pReader); -bool tqNextDataBlockFilterOut2(STqReader *pReader, SHashObj *filterOutUids); -int32_t tqRetrieveDataBlock2(SSDataBlock *pBlock, STqReader *pReader, SSubmitTbData **pSubmitTbDataRet); +bool tqNextBlockImpl(STqReader *pReader); +bool tqNextDataBlockFilterOut(STqReader *pReader, SHashObj *filterOutUids); +int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, STqReader *pReader, SSubmitTbData **pSubmitTbDataRet); int32_t tqRetrieveTaosxBlock2(STqReader *pReader, SArray *blocks, SArray *schemas, SSubmitTbData **pSubmitTbDataRet); -// int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, STqReader *pReader); -// int32_t tqRetrieveTaosxBlock(STqReader *pReader, SArray *blocks, SArray *schemas); int32_t vnodeEnqueueStreamMsg(SVnode *pVnode, SRpcMsg *pMsg); diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index ce987ca88e..69b3f9c3e0 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -684,7 +684,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma } if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); smaError("vgId:%d, process submit req for rsma suid:%" PRIu64 ", uid:%" PRIu64 " level %" PRIi8 " failed since %s", @@ -696,7 +696,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version); if (pReq) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); } } diff --git a/source/dnode/vnode/src/sma/smaTimeRange.c b/source/dnode/vnode/src/sma/smaTimeRange.c index 5058a7fc76..2a26f65bf9 100644 --- a/source/dnode/vnode/src/sma/smaTimeRange.c +++ b/source/dnode/vnode/src/sma/smaTimeRange.c @@ -332,7 +332,7 @@ _end: taosArrayDestroy(tagArray); taosArrayDestroy(pVals); if (pReq) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); } diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index 2cda12c0e1..0e9caf24ae 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -288,7 +288,7 @@ void tqCloseReader(STqReader* pReader) { } // free hash taosHashCleanup(pReader->tbIdHash); - tDestroySSubmitReq2(&pReader->submit, TSDB_MSG_FLG_DECODE); + tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); taosMemoryFree(pReader); } @@ -322,12 +322,11 @@ int32_t extractSubmitMsgFromWal(SWalReader* pReader, SPackedData* pPackedData) { return 0; } -void tqNextBlock(STqReader* pReader, SFetchRet* ret) { +int32_t tqNextBlock(STqReader* pReader, SSDataBlock* pBlock) { while (1) { if (pReader->msg2.msgStr == NULL) { if (walNextValidMsg(pReader->pWalReader) < 0) { - ret->fetchType = FETCH_TYPE__NONE; - return; + return FETCH_TYPE__NONE; } void* pBody = POINTER_SHIFT(pReader->pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg)); @@ -337,15 +336,14 @@ void tqNextBlock(STqReader* pReader, SFetchRet* ret) { tqReaderSetSubmitMsg(pReader, pBody, bodyLen, ver); } - while (tqNextDataBlock(pReader)) { - memset(&ret->data, 0, sizeof(SSDataBlock)); - int32_t code = tqRetrieveDataBlock2(&ret->data, pReader, NULL); - if (code != 0 || ret->data.info.rows == 0) { + while (tqNextBlockImpl(pReader)) { + memset(pBlock, 0, sizeof(SSDataBlock)); + int32_t code = tqRetrieveDataBlock(pBlock, pReader, NULL); + if (code != TSDB_CODE_SUCCESS || pBlock->info.rows == 0) { continue; } - ret->fetchType = FETCH_TYPE__DATA; - return; + return FETCH_TYPE__DATA; } } } @@ -367,7 +365,7 @@ int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, i return 0; } -bool tqNextDataBlock(STqReader* pReader) { +bool tqNextBlockImpl(STqReader* pReader) { if (pReader->msg2.msgStr == NULL) { return false; } @@ -387,20 +385,20 @@ bool tqNextDataBlock(STqReader* pReader) { tqDebug("tq reader block found, ver:%"PRId64", uid:%"PRId64, pReader->msg2.ver, pSubmitTbData->uid); return true; } else { - tqDebug("tq reader discard block, uid:%"PRId64", continue", pSubmitTbData->uid); + tqDebug("tq reader discard submit block, uid:%"PRId64", continue", pSubmitTbData->uid); } pReader->nextBlk++; } - tDestroySSubmitReq2(&pReader->submit, TSDB_MSG_FLG_DECODE); + tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); pReader->nextBlk = 0; pReader->msg2.msgStr = NULL; return false; } -bool tqNextDataBlockFilterOut2(STqReader* pReader, SHashObj* filterOutUids) { +bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) { if (pReader->msg2.msgStr == NULL) return false; int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData); @@ -415,7 +413,7 @@ bool tqNextDataBlockFilterOut2(STqReader* pReader, SHashObj* filterOutUids) { pReader->nextBlk++; } - tDestroySSubmitReq2(&pReader->submit, TSDB_MSG_FLG_DECODE); + tDestroySSubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); pReader->nextBlk = 0; pReader->msg2.msgStr = NULL; @@ -451,7 +449,7 @@ int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrap return 0; } -int32_t tqRetrieveDataBlock2(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbData** pSubmitTbDataRet) { +int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbData** pSubmitTbDataRet) { tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg2.msgStr, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); pReader->nextBlk++; @@ -560,7 +558,7 @@ int32_t tqRetrieveDataBlock2(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbD int32_t sourceIdx = 0; while (targetIdx < colActual) { if(sourceIdx >= numOfCols){ - tqError("tqRetrieveDataBlock2 sourceIdx:%d >= numOfCols:%d", sourceIdx, numOfCols); + tqError("tqRetrieveDataBlock sourceIdx:%d >= numOfCols:%d", sourceIdx, numOfCols); goto FAIL; } SColData* pCol = taosArrayGet(pCols, sourceIdx); @@ -568,7 +566,7 @@ int32_t tqRetrieveDataBlock2(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbD SColVal colVal; if(pCol->nVal != numOfRows){ - tqError("tqRetrieveDataBlock2 pCol->nVal:%d != numOfRows:%d", pCol->nVal, numOfRows); + tqError("tqRetrieveDataBlock pCol->nVal:%d != numOfRows:%d", pCol->nVal, numOfRows); goto FAIL; } diff --git a/source/dnode/vnode/src/tq/tqScan.c b/source/dnode/vnode/src/tq/tqScan.c index 1c916d9adb..e049e2d390 100644 --- a/source/dnode/vnode/src/tq/tqScan.c +++ b/source/dnode/vnode/src/tq/tqScan.c @@ -203,7 +203,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR if (pExec->subType == TOPIC_SUB_TYPE__TABLE) { STqReader* pReader = pExec->pTqReader; tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver); - while (tqNextDataBlock(pReader)) { + while (tqNextBlockImpl(pReader)) { taosArrayClear(pBlocks); taosArrayClear(pSchemas); SSubmitTbData* pSubmitTbDataRet = NULL; @@ -262,7 +262,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR } else if (pExec->subType == TOPIC_SUB_TYPE__DB) { STqReader* pReader = pExec->pTqReader; tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver); - while (tqNextDataBlockFilterOut2(pReader, pExec->execDb.pFilterOutTbUid)) { + while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) { taosArrayClear(pBlocks); taosArrayClear(pSchemas); SSubmitTbData* pSubmitTbDataRet = NULL; diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c index 62b81305b7..c2e6946b04 100644 --- a/source/dnode/vnode/src/tq/tqSink.c +++ b/source/dnode/vnode/src/tq/tqSink.c @@ -695,7 +695,7 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* len += sizeof(SSubmitReq2Msg); pBuf = rpcMallocCont(len); if (NULL == pBuf) { - tDestroySSubmitReq2(&submitReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); goto _end; } ((SSubmitReq2Msg*)pBuf)->header.vgId = TD_VID(pVnode); @@ -707,11 +707,11 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* tqError("failed to encode submit req since %s", terrstr()); tEncoderClear(&encoder); rpcFreeCont(pBuf); - tDestroySSubmitReq2(&submitReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); continue; } tEncoderClear(&encoder); - tDestroySSubmitReq2(&submitReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE); SRpcMsg msg = { .msgType = TDMT_VND_SUBMIT, diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 0dfde8f579..2e6d452e95 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -1388,7 +1388,7 @@ _exit: // clear taosArrayDestroy(newTbUids); - tDestroySSubmitReq2(pSubmitReq, 0 == pMsg->version ? TSDB_MSG_FLG_CMPT : TSDB_MSG_FLG_DECODE); + tDestroySSubmitReq(pSubmitReq, 0 == pMsg->version ? TSDB_MSG_FLG_CMPT : TSDB_MSG_FLG_DECODE); tDestroySSubmitRsp2(pSubmitRsp, TSDB_MSG_FLG_ENCODE); if (code) terrno = code; diff --git a/source/libs/executor/src/dataInserter.c b/source/libs/executor/src/dataInserter.c index 90d740bebd..22f388d406 100644 --- a/source/libs/executor/src/dataInserter.c +++ b/source/libs/executor/src/dataInserter.c @@ -301,7 +301,7 @@ _end: if (terrno != 0) { *ppReq = NULL; if (pReq) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); } return terrno; @@ -326,7 +326,7 @@ int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32 code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, uid, vgId, suid); if (code) { if (pReq) { - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); } @@ -335,7 +335,7 @@ int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32 } code = submitReqToMsg(vgId, pReq, pMsg, msgLen); - tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pReq, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pReq); return code; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 2389c7252e..8542481419 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1646,10 +1646,10 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { blockDataCleanup(pInfo->pRes); SDataBlockInfo* pBlockInfo = &pInfo->pRes->info; - while (tqNextDataBlock(pInfo->tqReader)) { + while (tqNextBlockImpl(pInfo->tqReader)) { SSDataBlock block = {0}; - int32_t code = tqRetrieveDataBlock2(&block, pInfo->tqReader, NULL); + int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader, NULL); if (code != TSDB_CODE_SUCCESS || block.info.rows == 0) { continue; } @@ -1687,23 +1687,23 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) { while (1) { - SFetchRet ret = {0}; - tqNextBlock(pInfo->tqReader, &ret); - tqOffsetResetToLog( - &pTaskInfo->streamInfo.currentOffset, - pInfo->tqReader->pWalReader->curVersion - 1); // curVersion move to next, so currentOffset = curVersion - 1 - - if (ret.fetchType == FETCH_TYPE__DATA) { - qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, ret.data.info.rows, + SSDataBlock block = {0}; + int32_t type = tqNextBlock(pInfo->tqReader, &block); + + // curVersion move to next, so currentOffset = curVersion - 1 + tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pInfo->tqReader->pWalReader->curVersion - 1); + + if (type == FETCH_TYPE__DATA) { + qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, block.info.rows, pTaskInfo->streamInfo.currentOffset.version); blockDataCleanup(pInfo->pRes); - setBlockIntoRes(pInfo, &ret.data, true); + setBlockIntoRes(pInfo, &block, true); if (pInfo->pRes->info.rows > 0) { qDebug("doQueueScan get data from log %" PRId64 " rows, return, version:%" PRId64, pInfo->pRes->info.rows, pTaskInfo->streamInfo.currentOffset.version); return pInfo->pRes; } - } else if (ret.fetchType == FETCH_TYPE__NONE) { + } else if (type == FETCH_TYPE__NONE) { qDebug("doQueueScan get none from log, return, version:%" PRId64, pTaskInfo->streamInfo.currentOffset.version); return NULL; } @@ -2072,11 +2072,10 @@ FETCH_NEXT_BLOCK: blockDataCleanup(pInfo->pRes); - while (tqNextDataBlock(pInfo->tqReader)) { + while (tqNextBlockImpl(pInfo->tqReader)) { SSDataBlock block = {0}; - int32_t code = tqRetrieveDataBlock2(&block, pInfo->tqReader, NULL); - + int32_t code = tqRetrieveDataBlock(&block, pInfo->tqReader, NULL); if (code != TSDB_CODE_SUCCESS || block.info.rows == 0) { continue; } diff --git a/source/libs/parser/src/parInsertUtil.c b/source/libs/parser/src/parInsertUtil.c index ac504b9809..a3b067b94d 100644 --- a/source/libs/parser/src/parInsertUtil.c +++ b/source/libs/parser/src/parInsertUtil.c @@ -324,7 +324,7 @@ void insDestroyVgroupDataCxt(SVgroupDataCxt* pVgCxt) { return; } - tDestroySSubmitReq2(pVgCxt->pData, TSDB_MSG_FLG_ENCODE); + tDestroySSubmitReq(pVgCxt->pData, TSDB_MSG_FLG_ENCODE); taosMemoryFree(pVgCxt->pData); taosMemoryFree(pVgCxt); } -- GitLab