From f1bd829977b43a0bbb12b4f9acdfb13d57f4a87d Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 2 Jun 2023 13:10:54 +0800 Subject: [PATCH] refactor: do some internal refactor. --- source/dnode/vnode/inc/vnode.h | 2 +- source/dnode/vnode/src/tq/tq.c | 3 ++- source/dnode/vnode/src/tq/tqMeta.c | 3 ++- source/dnode/vnode/src/tq/tqRead.c | 19 +++++++++++-------- source/libs/executor/src/executor.c | 2 +- source/libs/executor/src/scanoperator.c | 5 +++-- source/libs/stream/src/streamDispatch.c | 2 +- source/libs/stream/src/streamExec.c | 4 ++-- 8 files changed, 23 insertions(+), 17 deletions(-) diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index fbe2f4809c..8003419ec9 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -200,7 +200,7 @@ STqReader *tqReaderOpen(SVnode *pVnode); void tqReaderClose(STqReader *); void tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList); -int32_t tqReaderSetTbUidList(STqReader *pReader, const SArray *tbUidList); +int32_t tqReaderSetTbUidList(STqReader *pReader, const SArray *tbUidList, const char* id); int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList); int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 5e7e21e7bd..a6bed26884 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -708,8 +708,9 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); tqDebug("vgId:%d, idx %d, uid:%" PRId64, vgId, i, tbUid); } + pHandle->execHandle.pTqReader = tqReaderOpen(pVnode); - tqReaderSetTbUidList(pHandle->execHandle.pTqReader, tbUidList); + tqReaderSetTbUidList(pHandle->execHandle.pTqReader, tbUidList, NULL); taosArrayDestroy(tbUidList); buildSnapContext(handle.vnode, handle.version, req.suid, pHandle->execHandle.subType, pHandle->fetchMeta, diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index be5b5706e2..a720e556e6 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -344,8 +344,9 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); tqDebug("vgId:%d, idx %d, uid:%" PRId64, vgId, i, tbUid); } + handle.execHandle.pTqReader = tqReaderOpen(pTq->pVnode); - tqReaderSetTbUidList(handle.execHandle.pTqReader, tbUidList); + tqReaderSetTbUidList(handle.execHandle.pTqReader, tbUidList, NULL); taosArrayDestroy(tbUidList); buildSnapContext(reader.vnode, reader.version, handle.execHandle.execTb.suid, handle.execHandle.subType, diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index dbb7d564f8..f15a9898b5 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -394,8 +394,8 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id) { SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); if (pReader->tbIdHash == NULL) { - SSDataBlock* pRes = NULL; - int32_t code = tqRetrieveDataBlock(pReader, &pRes, NULL); + SSDataBlock* pRes = NULL; + int32_t code = tqRetrieveDataBlock(pReader, &pRes, NULL); if (code == TSDB_CODE_SUCCESS && pRes->info.rows > 0) { return true; } @@ -457,7 +457,7 @@ bool tqNextBlockImpl(STqReader* pReader, const char* idstr) { int32_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData); while (pReader->nextBlk < numOfBlocks) { - tqDebug("tq reader next data block, len:%d ver:%" PRId64 " index:%d/%d, %s", pReader->msg.msgLen, pReader->msg.ver, + tqDebug("try next data block, len:%d ver:%" PRId64 " index:%d/%d, %s", pReader->msg.msgLen, pReader->msg.ver, pReader->nextBlk, numOfBlocks, idstr); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); @@ -467,10 +467,11 @@ bool tqNextBlockImpl(STqReader* pReader, const char* idstr) { void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t)); if (ret != NULL) { - tqDebug("tq reader block found, ver:%" PRId64 ", uid:%" PRId64, pReader->msg.ver, pSubmitTbData->uid); + tqDebug("block found, ver:%" PRId64 ", uid:%" PRId64", %s", pReader->msg.ver, pSubmitTbData->uid, idstr); return true; } else { - tqDebug("tq reader discard submit block, uid:%" PRId64 ", continue", pSubmitTbData->uid); + tqDebug("discard submit block, uid:%" PRId64 ", total queried tables:%d continue %s", pSubmitTbData->uid, + taosHashGetSize(pReader->tbIdHash), idstr); } pReader->nextBlk++; @@ -604,7 +605,6 @@ static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SCol int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* id) { tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk); - SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk++); SSDataBlock* pBlock = pReader->pResBlock; @@ -612,6 +612,7 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* blockDataCleanup(pBlock); + int32_t vgId = pReader->pWalReader->pWal->cfg.vgId; int32_t sversion = pSubmitTbData->sver; int64_t suid = pSubmitTbData->suid; int64_t uid = pSubmitTbData->uid; @@ -628,7 +629,7 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* if (pReader->pSchemaWrapper == NULL) { tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", uid:%" PRId64 "version %d, possibly dropped table", - pReader->pWalReader->pWal->cfg.vgId, suid, uid, pReader->cachedSchemaVer); + vgId, suid, uid, pReader->cachedSchemaVer); pReader->cachedSchemaSuid = 0; terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND; return -1; @@ -642,6 +643,7 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* if (blockDataGetNumOfCols(pBlock) == 0) { int32_t code = buildResSDataBlock(pReader->pResBlock, pReader->pSchemaWrapper, pReader->pColIdList); if (code != TSDB_CODE_SUCCESS) { + tqError("vgId:%d failed to build data block, code:%s", vgId, tstrerror(code)); return code; } } @@ -998,7 +1000,7 @@ FAIL: void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) { pReader->pColIdList = pColIdList; } -int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) { +int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList, const char* id) { if (pReader->tbIdHash) { taosHashClear(pReader->tbIdHash); } else { @@ -1015,6 +1017,7 @@ int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) { taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0); } + tqDebug("s-task:%s %d tables are set to be queried target table", id, (int32_t) taosArrayGetSize(tbUidList)); return 0; } diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index ed8a4f3a3b..4d02362904 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -401,7 +401,7 @@ int32_t qUpdateTableListForStreamScanner(qTaskInfo_t tinfo, const SArray* tableI int32_t code = 0; if (isAdd) { - qDebug("add %d tables id into query list, %s", (int32_t)taosArrayGetSize(tableIdList), id); + qDebug("try to add %d tables id into query list, %s", (int32_t)taosArrayGetSize(tableIdList), id); } // traverse to the stream scanner node to add this table id diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 80a7505c77..5f09b0c67a 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -2308,7 +2308,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys SArray* pColIds = NULL; SStreamScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamScanInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - SStorageAPI* pAPI = &pTaskInfo->storageAPI; + SStorageAPI* pAPI = &pTaskInfo->storageAPI; + const char* idstr = pTaskInfo->id.str; if (pInfo == NULL || pOperator == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -2419,7 +2420,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys // set the extract column id to streamHandle pAPI->tqReaderFn.tqReaderSetColIdList(pInfo->tqReader, pColIds); SArray* tableIdList = extractTableIdList(((STableScanInfo*)(pInfo->pTableScanOp->info))->base.pTableListInfo); - code = pAPI->tqReaderFn.tqReaderSetQueryTableList(pInfo->tqReader, tableIdList); + code = pAPI->tqReaderFn.tqReaderSetQueryTableList(pInfo->tqReader, tableIdList, idstr); if (code != 0) { taosArrayDestroy(tableIdList); goto _error; diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index bd6a013de2..9501718416 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -318,7 +318,7 @@ int32_t doSendDispatchMsg(SStreamTask* pTask, const SStreamDispatchReq* pReq, in msg.pCont = buf; msg.msgType = pTask->dispatchMsgType; - qDebug("dispatch from s-task:%s to taskId:0x%x vgId:%d data msg", pTask->id.idStr, pReq->taskId, vgId); + qDebug("s-task:%s dispatch msg to taskId:0x%x vgId:%d data msg", pTask->id.idStr, pReq->taskId, vgId); tmsgSendReq(pEpSet, &msg); code = 0; diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 813bc50b87..c28cd6f334 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -133,7 +133,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i taosArrayPush(pRes, &block); - qDebug("s-task:%s (child %d) executed and get block, total blocks:%d, size:%.2fMiB", pTask->id.idStr, + qDebug("s-task:%s (child %d) executed and get %d result blocks, size:%.2fMiB", pTask->id.idStr, pTask->selfChildId, numOfBlocks, size / 1048576.0); // current output should be dispatched to down stream nodes @@ -371,7 +371,7 @@ int32_t streamExecForAll(SStreamTask* pTask) { streamQueueProcessSuccess(pTask->inputQueue); if (batchSize > MAX_STREAM_EXEC_BATCH_NUM) { - qDebug("s-task:%s maximum batch limit:%d reached, processing this batch of blocks", id, + qDebug("s-task:%s batch size limit:%d reached, start to process blocks", id, MAX_STREAM_EXEC_BATCH_NUM); break; }