xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: display the time window for force_window_close.

This commit is contained in:
Haojun Liao 2024-11-29 13:47:20 +08:00
parent 6580e7751f
commit 9a9a1828c1
5 changed files with 43 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/mnode/impl/inc/mndStream.h
View File

@ -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);

5
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -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;

10
source/dnode/mnode/impl/src/mndStream.c
View File

@ -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++;
}

22
source/dnode/mnode/impl/src/mndStreamUtil.c
View File

@ -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));
}

30
source/libs/stream/src/streamHb.c
View File

@ -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);