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

fix:stream crash && add debug log

This commit is contained in:
54liuyao 2022-12-01 14:15:50 +08:00
parent ce768920ef
commit bccb31aced
4 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/executor/inc/executorimpl.h
View File

@ -536,6 +536,7 @@ typedef struct SStreamIntervalOperatorInfo {
SArray* pChildren; SArray* pChildren;
SStreamState* pState; SStreamState* pState;
SWinKey delKey; SWinKey delKey;
uint64_t numOfDatapack;
} SStreamIntervalOperatorInfo; } SStreamIntervalOperatorInfo;
typedef struct SDataGroupInfo { typedef struct SDataGroupInfo {

4
source/libs/executor/src/filloperator.c
View File

@ -1146,7 +1146,7 @@ static void doDeleteFillResult(SOperatorInfo* pOperator) {
if (delTs > nextKey.ts) { if (delTs > nextKey.ts) {
break; break;
} }
endTs = delTs;
SWinKey delKey = {.groupId = delGroupId, .ts = delTs}; SWinKey delKey = {.groupId = delGroupId, .ts = delTs};
if (delTs == nextKey.ts) { if (delTs == nextKey.ts) {
code = streamStateCurNext(pOperator->pTaskInfo->streamInfo.pState, pCur); code = streamStateCurNext(pOperator->pTaskInfo->streamInfo.pState, pCur);
@ -1159,7 +1159,7 @@ static void doDeleteFillResult(SOperatorInfo* pOperator) {
streamStateFreeCur(pCur); streamStateFreeCur(pCur);
pCur = streamStateGetAndCheckCur(pOperator->pTaskInfo->streamInfo.pState, &nextKey); pCur = streamStateGetAndCheckCur(pOperator->pTaskInfo->streamInfo.pState, &nextKey);
} }
endTs = TMAX(ts, nextKey.ts - 1); endTs = TMAX(delTs, nextKey.ts - 1);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
break; break;
} }

2
source/libs/executor/src/scanoperator.c
View File

@ -1113,7 +1113,7 @@ static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, uint64_t
if (hasGroup) { if (hasGroup) {
(*pRowIndex) += 1; (*pRowIndex) += 1;
} else { } else {
while ((groupId == gpIdCol[(*pRowIndex)] && startTsCol[*pRowIndex] < endWin.ekey)) { while ((groupId == gpIdCol[(*pRowIndex)] && startTsCol[*pRowIndex] <= endWin.ekey)) {
(*pRowIndex) += 1; (*pRowIndex) += 1;
if ((*pRowIndex) == pDataBlockInfo->rows) { if ((*pRowIndex) == pDataBlockInfo->rows) {
break; break;

9
source/libs/executor/src/timewindowoperator.c
View File

@ -2516,9 +2516,11 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream); SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) { if (pBlock == NULL) {
pOperator->status = OP_RES_TO_RETURN; pOperator->status = OP_RES_TO_RETURN;
qDebug("%s return data", IS_FINAL_OP(pInfo) ? "interval final" : "interval semi"); qDebug("===stream===return data:%s. recv datablock num:%" PRIu64 , IS_FINAL_OP(pInfo) ? "interval final" : "interval semi", pInfo->numOfDatapack);
pInfo->numOfDatapack = 0;
break; break;
} }
pInfo->numOfDatapack++;
printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval final recv" : "interval semi recv"); printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval final recv" : "interval semi recv");
ASSERT(pBlock->info.type != STREAM_INVERT); ASSERT(pBlock->info.type != STREAM_INVERT);
@ -2734,6 +2736,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pInfo->pDelWins = taosArrayInit(4, sizeof(SWinKey)); pInfo->pDelWins = taosArrayInit(4, sizeof(SWinKey));
pInfo->delKey.ts = INT64_MAX; pInfo->delKey.ts = INT64_MAX;
pInfo->delKey.groupId = 0; pInfo->delKey.groupId = 0;
pInfo->numOfDatapack = 0;
pOperator->operatorType = pPhyNode->type; pOperator->operatorType = pPhyNode->type;
pOperator->blocking = true; pOperator->blocking = true;
@ -4700,8 +4703,11 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
while (1) { while (1) {
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream); SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) { if (pBlock == NULL) {
qDebug("===stream===return data:single interval. recv datablock num:%" PRIu64, pInfo->numOfDatapack);
pInfo->numOfDatapack = 0;
break; break;
} }
pInfo->numOfDatapack++;
printDataBlock(pBlock, "single interval recv"); printDataBlock(pBlock, "single interval recv");
if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT || if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
@ -4851,6 +4857,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->pChildren = NULL; pInfo->pChildren = NULL;
pInfo->delKey.ts = INT64_MAX; pInfo->delKey.ts = INT64_MAX;
pInfo->delKey.groupId = 0; pInfo->delKey.groupId = 0;
pInfo->numOfDatapack = 0;
setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED, setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED,
pInfo, pTaskInfo); pInfo, pTaskInfo);