From 9a9a1828c123ab1d7c63ef36095f04f528dc4bd4 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 29 Nov 2024 13:47:20 +0800 Subject: [PATCH] refactor: display the time window for force_window_close. --- source/dnode/mnode/impl/inc/mndStream.h | 2 +- source/dnode/mnode/impl/src/mndScheduler.c | 5 ++-- source/dnode/mnode/impl/src/mndStream.c | 10 +++++-- source/dnode/mnode/impl/src/mndStreamUtil.c | 22 +++++++-------- source/libs/stream/src/streamHb.c | 30 ++++++++++++++------- 5 files changed, 43 insertions(+), 26 deletions(-) diff --git a/source/dnode/mnode/impl/inc/mndStream.h b/source/dnode/mnode/impl/inc/mndStream.h index c9155f536c..b129cde72c 100644 --- a/source/dnode/mnode/impl/inc/mndStream.h +++ b/source/dnode/mnode/impl/inc/mndStream.h @@ -178,7 +178,7 @@ int64_t mndClearChkptReportInfo(SHashObj* pHash, int64_t streamId); int32_t mndResetChkptReportInfo(SHashObj* pHash, int64_t streamId); int32_t setStreamAttrInResBlock(SStreamObj *pStream, SSDataBlock *pBlock, int32_t numOfRows); -int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlock *pBlock, int32_t numOfRows); +int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlock *pBlock, int32_t nRows, int32_t p); int32_t mndProcessResetStatusReq(SRpcMsg *pReq); diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index e67e4a963b..f48e4531de 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -404,8 +404,7 @@ static int32_t doAddSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStre return code; } - mDebug("doAddSourceTask taskId:%s, %p vgId:%d, isFillHistory:%d", pTask->id.idStr, pTask, pVgroup->vgId, - isHistoryTask); + mDebug("doAddSourceTask taskId:%s, %p vgId:%d, historyTask:%d", pTask->id.idStr, pTask, pVgroup->vgId, isHistoryTask); if (pStream->conf.fillHistory) { haltInitialTaskStatus(pTask, plan, isHistoryTask); @@ -461,7 +460,7 @@ static int32_t addSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream addNewTaskList(pStream); while (1) { - SVgObj* pVgroup; + SVgObj* pVgroup = NULL; pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); if (pIter == NULL) { break; diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 6336cd6e49..5d41e1506c 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -21,7 +21,6 @@ #include "mndShow.h" #include "mndStb.h" #include "mndTrans.h" -#include "mndVgroup.h" #include "osMemory.h" #include "parser.h" #include "taoserror.h" @@ -1610,6 +1609,13 @@ static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock } } + int32_t precision = TSDB_TIME_PRECISION_MILLI; + SDbObj *pSourceDb = mndAcquireDb(pMnode, pStream->sourceDb); + if (pSourceDb != NULL) { + precision = pSourceDb->cfg.precision; + mndReleaseDb(pMnode, pSourceDb); + } + // add row for each task SStreamTaskIter *pIter = NULL; code = createStreamTaskIter(pStream, &pIter); @@ -1628,7 +1634,7 @@ static int32_t mndRetrieveStreamTask(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock break; } - code = setTaskAttrInResBlock(pStream, pTask, pBlock, numOfRows); + code = setTaskAttrInResBlock(pStream, pTask, pBlock, numOfRows, precision); if (code == TSDB_CODE_SUCCESS) { numOfRows++; } diff --git a/source/dnode/mnode/impl/src/mndStreamUtil.c b/source/dnode/mnode/impl/src/mndStreamUtil.c index f9b7644af4..ad2f71a064 100644 --- a/source/dnode/mnode/impl/src/mndStreamUtil.c +++ b/source/dnode/mnode/impl/src/mndStreamUtil.c @@ -1043,7 +1043,7 @@ _end: return code; } -int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlock *pBlock, int32_t numOfRows) { +int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlock *pBlock, int32_t numOfRows, int32_t precision) { SColumnInfoData *pColInfo = NULL; int32_t cols = 0; int32_t code = 0; @@ -1103,14 +1103,11 @@ int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlo // level char level[20 + VARSTR_HEADER_SIZE] = {0}; if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - memcpy(varDataVal(level), "source", 6); - varDataSetLen(level, 6); + STR_WITH_SIZE_TO_VARSTR(level, "source", 6); } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) { - memcpy(varDataVal(level), "agg", 3); - varDataSetLen(level, 3); + STR_WITH_SIZE_TO_VARSTR(level, "agg", 3); } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) { - memcpy(varDataVal(level), "sink", 4); - varDataSetLen(level, 4); + STR_WITH_SIZE_TO_VARSTR(level, "sink", 4); } pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); @@ -1234,10 +1231,13 @@ int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlo if (pTask->info.taskLevel == TASK_LEVEL__SINK) { const char *sinkStr = "%.2f MiB"; snprintf(buf, tListLen(buf), sinkStr, pe->sinkDataSize); - } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { - // offset info - const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; - snprintf(buf, tListLen(buf), offsetStr, pe->processedVer, pe->verRange.minVer, pe->verRange.maxVer); + } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { // offset info + if (pTask->info.trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) { + taosFormatUtcTime(buf, tListLen(buf), pe->processedVer, precision); + } else { + const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]"; + snprintf(buf, tListLen(buf), offsetStr, pe->processedVer, pe->verRange.minVer, pe->verRange.maxVer); + } } else { memset(buf, 0, tListLen(buf)); } diff --git a/source/libs/stream/src/streamHb.c b/source/libs/stream/src/streamHb.c index 25cb28f77c..3a4e3eef89 100644 --- a/source/libs/stream/src/streamHb.c +++ b/source/libs/stream/src/streamHb.c @@ -75,6 +75,25 @@ static void addUpdateNodeIntoHbMsg(SStreamTask* pTask, SStreamHbMsg* pMsg) { streamMutexUnlock(&pTask->lock); } +static void setProcessProgress(SStreamTask* pTask, STaskStatusEntry* pEntry) { + if (pTask->info.taskLevel != TASK_LEVEL__SOURCE) { + return; + } + + if (pTask->info.trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) { + pEntry->processedVer = pTask->status.latestForceWindow.skey; + } else { + if (pTask->exec.pWalReader != NULL) { + pEntry->processedVer = walReaderGetCurrentVer(pTask->exec.pWalReader) - 1; + if (pEntry->processedVer < 0) { + pEntry->processedVer = pTask->chkInfo.processedVer; + } + + walReaderValidVersionRange(pTask->exec.pWalReader, &pEntry->verRange.minVer, &pEntry->verRange.maxVer); + } + } +} + static int32_t doSendHbMsgInfo(SStreamHbMsg* pMsg, SStreamMeta* pMeta, SEpSet* pEpset) { int32_t code = 0; int32_t tlen = 0; @@ -209,16 +228,9 @@ int32_t streamMetaSendHbHelper(SStreamMeta* pMeta) { } streamMutexUnlock(&pTask->lock); - if (pTask->exec.pWalReader != NULL) { - entry.processedVer = walReaderGetCurrentVer(pTask->exec.pWalReader) - 1; - if (entry.processedVer < 0) { - entry.processedVer = pTask->chkInfo.processedVer; - } - - walReaderValidVersionRange(pTask->exec.pWalReader, &entry.verRange.minVer, &entry.verRange.maxVer); - } - + setProcessProgress(pTask, &entry); addUpdateNodeIntoHbMsg(pTask, pMsg); + p = taosArrayPush(pMsg->pTaskStatus, &entry); if (p == NULL) { stError("failed to add taskInfo:0x%x in hbMsg, vgId:%d", pTask->id.taskId, pMeta->vgId);