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

fix(stream): check return value.

This commit is contained in:
Haojun Liao 2024-07-23 16:38:05 +08:00
parent 7497f193c7
commit cd0722fe1f
2 changed files with 45 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
source/libs/executor/inc/tsort.h
View File

@ -165,13 +165,13 @@ void tsortGetValue(STupleHandle* pVHandle, int32_t colId, void** pVal);
* @return * @return
*/ */
uint64_t tsortGetGroupId(STupleHandle* pVHandle); uint64_t tsortGetGroupId(STupleHandle* pVHandle);
void* tsortGetBlockInfo(STupleHandle* pVHandle); void tsortGetBlockInfo(STupleHandle* pVHandle, SDataBlockInfo* pInfo);
/** /**
* *
* @param pSortHandle * @param pSortHandle
* @return * @return
*/ */
SSDataBlock* tsortGetSortedDataBlock(const SSortHandle* pSortHandle); int32_t tsortGetSortedDataBlock(const SSortHandle* pSortHandle, SSDataBlock** pBlock);
/** /**
* return the sort execution information. * return the sort execution information.
@ -215,8 +215,8 @@ int32_t tsortCompAndBuildKeys(const SArray* pSortCols, char* keyBuf, int32_t* ke
*/ */
void tsortSetMergeLimitReachedFp(SSortHandle* pHandle, void (*mergeLimitReached)(uint64_t tableUid, void* param), void* param); void tsortSetMergeLimitReachedFp(SSortHandle* pHandle, void (*mergeLimitReached)(uint64_t tableUid, void* param), void* param);
int tsortComparBlockCell(SSDataBlock* pLeftBlock, SSDataBlock* pRightBlock, int tsortComparBlockCell(SSDataBlock* pLeftBlock, SSDataBlock* pRightBlock, int32_t leftRowIndex, int32_t rightRowIndex,
int32_t leftRowIndex, int32_t rightRowIndex, void* pOrder); void* pOrder);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

53
source/libs/executor/src/tsort.c
View File

@ -167,8 +167,17 @@ static void* tupleGetField(char* t, uint32_t colIdx, uint32_t colNum) {
return t + *tupleOffset(t, colIdx); return t + *tupleOffset(t, colIdx);
} }
SSDataBlock* tsortGetSortedDataBlock(const SSortHandle* pSortHandle) { int32_t tsortGetSortedDataBlock(const SSortHandle* pSortHandle, SSDataBlock** pBlock) {
return createOneDataBlock(pSortHandle->pDataBlock, false); if (pBlock == NULL) {
return TSDB_CODE_INVALID_PARA;
}
*pBlock = createOneDataBlock(pSortHandle->pDataBlock, false);
if (*pBlock == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
} else {
return TSDB_CODE_SUCCESS;
}
} }
#define AllocatedTupleType 0 #define AllocatedTupleType 0
@ -1099,7 +1108,11 @@ int32_t tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupl
} }
pBlock->info.dataLoad = 1; pBlock->info.dataLoad = 1;
pBlock->info.scanFlag = ((SDataBlockInfo*)tsortGetBlockInfo(pTupleHandle))->scanFlag;
SDataBlockInfo info = {0};
tsortGetBlockInfo(pTupleHandle, &info);
pBlock->info.scanFlag = info.scanFlag;
pBlock->info.rows += 1; pBlock->info.rows += 1;
} else { } else {
@ -1121,7 +1134,10 @@ int32_t tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupl
} }
pBlock->info.dataLoad = 1; pBlock->info.dataLoad = 1;
pBlock->info.scanFlag = ((SDataBlockInfo*)tsortGetBlockInfo(pTupleHandle))->scanFlag; SDataBlockInfo info = {0};
tsortGetBlockInfo(pTupleHandle, &info);
pBlock->info.scanFlag = info.scanFlag;
pBlock->info.rows += 1; pBlock->info.rows += 1;
} }
@ -1463,35 +1479,41 @@ static int32_t initRowIdSort(SSortHandle* pHandle) {
SColumnInfoData pkCol = {0}; SColumnInfoData pkCol = {0};
SSDataBlock* pSortInput = createDataBlock(); SSDataBlock* pSortInput = createDataBlock();
pHandle->pDataBlock = pSortInput;
SColumnInfoData tsCol = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, 8, 1); SColumnInfoData tsCol = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, 8, 1);
int32_t code = blockDataAppendColInfo(pSortInput, &tsCol); int32_t code = blockDataAppendColInfo(pSortInput, &tsCol);
if (code) { if (code) {
blockDataDestroy(pSortInput);
return code; return code;
} }
SColumnInfoData regionIdCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 2); SColumnInfoData regionIdCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 2);
code = blockDataAppendColInfo(pSortInput, &regionIdCol); code = blockDataAppendColInfo(pSortInput, &regionIdCol);
if (code) { if (code) {
blockDataDestroy(pSortInput);
return code; return code;
} }
SColumnInfoData offsetCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 3); SColumnInfoData offsetCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 3);
code = blockDataAppendColInfo(pSortInput, &offsetCol); code = blockDataAppendColInfo(pSortInput, &offsetCol);
if (code) { if (code) {
blockDataDestroy(pSortInput);
return code; return code;
} }
SColumnInfoData lengthCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 4); SColumnInfoData lengthCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 4);
code = blockDataAppendColInfo(pSortInput, &lengthCol); code = blockDataAppendColInfo(pSortInput, &lengthCol);
if (code) { if (code) {
blockDataDestroy(pSortInput);
return code; return code;
} }
if (pHandle->bSortPk) { if (pHandle->bSortPk) {
pkCol = createColumnInfoData(extPkCol->info.type, extPkCol->info.bytes, 5); pkCol = createColumnInfoData(extPkCol->info.type, extPkCol->info.bytes, 5);
code = blockDataAppendColInfo(pSortInput, &pkCol); code = blockDataAppendColInfo(pSortInput, &pkCol);
if (code) {
blockDataDestroy(pSortInput);
return code;
}
} }
blockDataDestroy(pHandle->pDataBlock); blockDataDestroy(pHandle->pDataBlock);
@ -1990,12 +2012,13 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
bool bExtractedBlock = false; bool bExtractedBlock = false;
bool bSkipBlock = false; bool bSkipBlock = false;
if (pBlk != NULL && pHandle->mergeLimit > 0) { if (pBlk != NULL && pHandle->mergeLimit > 0) {
pBlk = NULL; SSDataBlock* p = NULL;
code = getRowsBlockWithinMergeLimit(pHandle, mTableNumRows, pBlk, &bExtractedBlock, &bSkipBlock, &p);
code = getRowsBlockWithinMergeLimit(pHandle, mTableNumRows, pBlk, &bExtractedBlock, &bSkipBlock, &pBlk);
if (bSkipBlock || code != 0) { if (bSkipBlock || code != 0) {
continue; continue;
} }
pBlk = p;
} }
if (pBlk != NULL) { if (pBlk != NULL) {
@ -2416,15 +2439,21 @@ static int32_t tsortOpenForPQSort(SSortHandle* pHandle) {
while (1) { while (1) {
// fetch data // fetch data
SSDataBlock* pBlock = pHandle->fetchfp(source->param); SSDataBlock* pBlock = pHandle->fetchfp(source->param);
if (NULL == pBlock) break; if (NULL == pBlock) {
break;
}
if (pHandle->beforeFp != NULL) { if (pHandle->beforeFp != NULL) {
pHandle->beforeFp(pBlock, pHandle->param); pHandle->beforeFp(pBlock, pHandle->param);
} }
if (pHandle->pDataBlock == NULL) { if (pHandle->pDataBlock == NULL) {
pHandle->pDataBlock = createOneDataBlock(pBlock, false); pHandle->pDataBlock = createOneDataBlock(pBlock, false);
} }
if (pHandle->pDataBlock == NULL) return TSDB_CODE_OUT_OF_MEMORY;
if (pHandle->pDataBlock == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
size_t colNum = blockDataGetNumOfCols(pBlock); size_t colNum = blockDataGetNumOfCols(pBlock);
@ -2581,7 +2610,7 @@ void tsortGetValue(STupleHandle* pVHandle, int32_t colIndex, void** pVal) {
} }
uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.id.groupId; } uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.id.groupId; }
void* tsortGetBlockInfo(STupleHandle* pVHandle) { return &pVHandle->pBlock->info; } void tsortGetBlockInfo(STupleHandle* pVHandle, SDataBlockInfo* pBlockInfo) { *pBlockInfo = pVHandle->pBlock->info; }
SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) { SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) {
SSortExecInfo info = {0}; SSortExecInfo info = {0};