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

enh(query):remove void for operator

This commit is contained in:
54liuyao 2024-09-04 18:37:41 +08:00
parent a9fe796484
commit d3e4203dcb
5 changed files with 53 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
source/libs/executor/src/exchangeoperator.c
View File

@ -464,7 +464,10 @@ _error:
void destroyExchangeOperatorInfo(void* param) {
SExchangeInfo* pExInfo = (SExchangeInfo*)param;
(void)taosRemoveRef(exchangeObjRefPool, pExInfo->self);
int32_t code = taosRemoveRef(exchangeObjRefPool, pExInfo->self);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
}
void freeBlock(void* pParam) {
@ -505,7 +508,10 @@ void doDestroyExchangeOperatorInfo(void* param) {
blockDataDestroy(pExInfo->pDummyBlock);
tSimpleHashCleanup(pExInfo->pHashSources);
(void)tsem_destroy(&pExInfo->ready);
int32_t code = tsem_destroy(&pExInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
taosMemoryFreeClear(pExInfo->pTaskId);
taosMemoryFreeClear(param);
@ -561,9 +567,13 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
if (code != TSDB_CODE_SUCCESS) {
code = TAOS_SYSTEM_ERROR(code);
qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
return code;
}
(void)taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
code = taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
return code;
}
@ -1190,7 +1200,14 @@ static int32_t exchangeWait(SOperatorInfo* pOperator, SExchangeInfo* pExchangeIn
return pTask->code;
}
}
(void)tsem_wait(&pExchangeInfo->ready);
code = tsem_wait(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
pTask->code = code;
return pTask->code;
}
if (pTask->pWorkerCb) {
code = pTask->pWorkerCb->afterRecoverFromBlocking(pTask->pWorkerCb->pPool);
if (code != TSDB_CODE_SUCCESS) {

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

@ -904,8 +904,14 @@ void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {
}
SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId);
if (pExchangeInfo) {
(void)tsem_post(&pExchangeInfo->ready);
(void)taosReleaseRef(exchangeObjRefPool, pStop->refId);
int32_t code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
code = taosReleaseRef(exchangeObjRefPool, pStop->refId);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
}
}

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

@ -671,7 +671,8 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
STableCachedVal* pVal = taosLRUCacheValue(pCache->pTableMetaEntryCache, h);
val = *pVal;
(void)taosLRUCacheRelease(pCache->pTableMetaEntryCache, h, false);
bool bRes = taosLRUCacheRelease(pCache->pTableMetaEntryCache, h, false);
qTrace("release LRU cache, res %d", bRes);
}
qDebug("retrieve table meta from cache:%" PRIu64 ", hit:%" PRIu64 " miss:%" PRIu64 ", %s", pCache->metaFetch,
@ -893,7 +894,10 @@ void markGroupProcessed(STableScanInfo* pInfo, uint64_t groupId) {
if (pInfo->base.pTableListInfo->groupOffset) {
pInfo->countState = TABLE_COUNT_STATE_PROCESSED;
} else {
(void)taosHashRemove(pInfo->base.pTableListInfo->remainGroups, &groupId, sizeof(groupId));
int32_t code = taosHashRemove(pInfo->base.pTableListInfo->remainGroups, &groupId, sizeof(groupId));
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
}
}

17
source/libs/executor/src/sysscanoperator.c
View File

@ -2178,7 +2178,12 @@ static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableSca
T_LONG_JMP(pTaskInfo->env, code);
}
(void)tsem_wait(&pInfo->ready);
code = tsem_wait(&pInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
if (pTaskInfo->code) {
qError("%s load meta data from mnode failed, totalRows:%" PRIu64 ", code:%s", GET_TASKID(pTaskInfo),
@ -2328,7 +2333,10 @@ void extractTbnameSlotId(SSysTableScanInfo* pInfo, const SScanPhysiNode* pScanNo
void destroySysScanOperator(void* param) {
SSysTableScanInfo* pInfo = (SSysTableScanInfo*)param;
(void)tsem_destroy(&pInfo->ready);
int32_t code = tsem_destroy(&pInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
blockDataDestroy(pInfo->pRes);
if (pInfo->name.type == TSDB_TABLE_NAME_T) {
@ -2384,7 +2392,10 @@ int32_t loadSysTableCallback(void* param, SDataBuf* pMsg, int32_t code) {
}
}
(void)tsem_post(&pScanResInfo->ready);
int32_t res = tsem_post(&pScanResInfo->ready);
if (res != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(res));
}
return TSDB_CODE_SUCCESS;
}

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

@ -1231,7 +1231,7 @@ void destroyIntervalOperatorInfo(void* param) {
cleanupAggSup(&pInfo->aggSup);
cleanupExprSupp(&pInfo->scalarSupp);
(void)tdListFree(pInfo->binfo.resultRowInfo.openWindow);
pInfo->binfo.resultRowInfo.openWindow = tdListFree(pInfo->binfo.resultRowInfo.openWindow);
taosArrayDestroy(pInfo->pInterpCols);
pInfo->pInterpCols = NULL;
@ -2132,7 +2132,7 @@ typedef struct SGroupTimeWindow {
void destroyMergeIntervalOperatorInfo(void* param) {
SMergeIntervalAggOperatorInfo* miaInfo = (SMergeIntervalAggOperatorInfo*)param;
(void)tdListFree(miaInfo->groupIntervals);
miaInfo->groupIntervals = tdListFree(miaInfo->groupIntervals);
destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo);
taosMemoryFreeClear(param);
@ -2162,7 +2162,8 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t
STimeWindow* prevWin = &prevGrpWin->window;
if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
(void)tdListPopNode(miaInfo->groupIntervals, listNode);
SListNode* tmp = tdListPopNode(miaInfo->groupIntervals, listNode);
taosMemoryFreeClear(tmp);
}
}