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

fix ci issue

This commit is contained in:
54liuyao 2024-10-09 11:32:20 +08:00
parent 08283a34c2
commit 58d8b9f84e
3 changed files with 24 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3400,7 +3400,10 @@ static bool isStreamWindow(SStreamScanInfo* pInfo) {
static int32_t copyGetResultBlock(SSDataBlock* dest, const SSDataBlock* src) {
TSKEY start = src->info.window.skey;
TSKEY end = src->info.window.ekey;
blockDataEnsureCapacity(dest, 1);
int32_t code = blockDataEnsureCapacity(dest, 1);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
return appendDataToSpecialBlock(dest, &start, &end, NULL, NULL, NULL);
}

8
source/libs/executor/src/streamfilloperator.c
View File

@ -1263,9 +1263,13 @@ SStreamFillInfo* initStreamFillInfo(SStreamFillSupporter* pFillSup, SSDataBlock*
if (pColData == NULL) {
SPoint dummy = {0};
dummy.val = taosMemoryCalloc(1, 1);
taosArrayPush(pFillInfo->pLinearInfo->pEndPoints, &dummy);
void* tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pEndPoints, &dummy);
QUERY_CHECK_NULL(tmpRes, code, lino, _end, terrno);
dummy.val = taosMemoryCalloc(1, 1);
taosArrayPush(pFillInfo->pLinearInfo->pNextEndPoints, &dummy);
tmpRes = taosArrayPush(pFillInfo->pLinearInfo->pNextEndPoints, &dummy);
QUERY_CHECK_NULL(tmpRes, code, lino, _end, terrno);
continue;
}
SPoint value = {0};

18
source/libs/stream/src/tstreamFileState.c
View File

@ -458,7 +458,8 @@ int32_t flushRowBuff(SStreamFileState* pFileState) {
}
if (pFileState->searchBuff) {
clearFlushedRowBuff(pFileState, pFlushList, pFileState->curRowCount, true);
code = clearFlushedRowBuff(pFileState, pFlushList, pFileState->curRowCount, true);
QUERY_CHECK_CODE(code, lino, _end);
}
flushSnapshot(pFileState, pFlushList, false);
@ -712,11 +713,13 @@ int32_t getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void**
goto _end;
}
recoverStateRowBuff(pFileState, pPos);
code = recoverStateRowBuff(pFileState, pPos);
QUERY_CHECK_CODE(code, lino, _end);
(*pVal) = pPos->pRowBuff;
if (!pPos->needFree) {
code = tdListPrepend(pFileState->usedBuffs, &pPos);
QUERY_CHECK_CODE(code, lino, _end);
}
_end:
@ -1023,11 +1026,13 @@ int32_t getFunctionRowBuff(SStreamFileState* pFileState, void* pKey, int32_t key
int32_t recoverFillSnapshot(SStreamFileState* pFileState, int64_t ckId) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (pFileState->maxTs != INT64_MIN) {
int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
? INT64_MIN
: pFileState->maxTs - pFileState->deleteMark;
deleteExpiredCheckPoint(pFileState, mark);
code = deleteExpiredCheckPoint(pFileState, mark);
QUERY_CHECK_CODE(code, lino, _end);
}
SStreamStateCur* pCur = streamStateFillSeekToLast_rocksdb(pFileState->pFileStore);
@ -1065,6 +1070,10 @@ int32_t recoverFillSnapshot(SStreamFileState* pFileState, int64_t ckId) {
}
streamStateFreeCur(pCur);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
@ -1082,7 +1091,8 @@ int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, voi
(*ppPos)->beFlushed = false;
(*pWinCode) = TSDB_CODE_SUCCESS;
if ((*ppPos)->pRowBuff == NULL) {
recoverStateRowBuff(pFileState, *ppPos);
code = recoverStateRowBuff(pFileState, *ppPos);
QUERY_CHECK_CODE(code, lino, _end);
}
goto _end;
}