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

delete some log

This commit is contained in:
liuyao 2023-04-12 17:39:44 +08:00
parent 9be89ce00a
commit e42caca9c2
2 changed files with 3 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2151,7 +2151,7 @@ FETCH_NEXT_BLOCK:
// record the scan action. // record the scan action.
pInfo->numOfExec++; pInfo->numOfExec++;
pOperator->resultInfo.totalRows += pBlockInfo->rows; pOperator->resultInfo.totalRows += pBlockInfo->rows;
printDataBlock(pInfo->pRes, "stream scan"); // printDataBlock(pInfo->pRes, "stream scan");
qDebug("scan rows: %" PRId64, pBlockInfo->rows); qDebug("scan rows: %" PRId64, pBlockInfo->rows);
if (pBlockInfo->rows > 0) { if (pBlockInfo->rows > 0) {

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

@ -2400,7 +2400,6 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
STimeWindow nextWin = {0}; STimeWindow nextWin = {0};
if (IS_FINAL_OP(pInfo)) { if (IS_FINAL_OP(pInfo)) {
nextWin = getFinalTimeWindow(ts, &pInfo->interval); nextWin = getFinalTimeWindow(ts, &pInfo->interval);
qDebug("===stream===final ts:%" PRId64 ", getFinalTimeWindow:%" PRId64,ts, nextWin.skey);
} else { } else {
nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, TSDB_ORDER_ASC); nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, TSDB_ORDER_ASC);
} }
@ -2428,7 +2427,6 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) { if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
int32_t size = taosArrayGetSize(pInfo->pChildren); int32_t size = taosArrayGetSize(pInfo->pChildren);
addPullWindow(pInfo->pPullDataMap, &winRes, size); addPullWindow(pInfo->pPullDataMap, &winRes, size);
qDebug("===stream===prepare retrive %" PRId64 ", size:%d", winRes.ts, size);
} }
} else { } else {
int32_t index = -1; int32_t index = -1;
@ -2445,17 +2443,14 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
} }
if (ignore) { if (ignore) {
// startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin); startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, startPos);
int32_t prevEndPos = startPos;
startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos);
if (startPos < 0) { if (startPos < 0) {
break; break;
} }
continue; continue;
} }
} }
qDebug("===stream===final setIntervalOutputBuf:%" PRId64, nextWin.skey);
int32_t code = setIntervalOutputBuf(pInfo->pState, &nextWin, &pResPos, groupId, pSup->pCtx, numOfOutput, int32_t code = setIntervalOutputBuf(pInfo->pState, &nextWin, &pResPos, groupId, pSup->pCtx, numOfOutput,
pSup->rowEntryInfoOffset, &pInfo->aggSup); pSup->rowEntryInfoOffset, &pInfo->aggSup);
pResult = (SResultRow*)pResPos->pRowBuff; pResult = (SResultRow*)pResPos->pRowBuff;
@ -2474,7 +2469,6 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
.groupId = groupId, .groupId = groupId,
}; };
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) { if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) {
qDebug("===stream===is final%d saveWinResult:%" PRId64, IS_FINAL_OP(pInfo), key.ts);
saveWinResult(&key, pResPos, pUpdatedMap); saveWinResult(&key, pResPos, pUpdatedMap);
} }