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

Merge branch '3.0' into enh/TD-30988-3.0

This commit is contained in:
kailixu 2024-07-25 17:31:28 +08:00
parent 90ff2b34ba a3b30192ae
commit ae2fa9e1ce
21 changed files with 1086 additions and 549 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/executor/inc/tfill.h
View File

@ -128,9 +128,9 @@ SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprIn
int32_t numOfNotFillCols, const struct SNodeListNode* val);
bool taosFillHasMoreResults(struct SFillInfo* pFillInfo);
SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t slotId,
int32_t order, const char* id, SExecTaskInfo* pTaskInfo);
void taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t slotId,
int32_t order, const char* id, SExecTaskInfo* pTaskInfo, SFillInfo** ppFillInfo);
void* taosDestroyFillInfo(struct SFillInfo* pFillInfo);
int32_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity);

15
source/libs/executor/src/aggregateoperator.c
View File

@ -252,14 +252,15 @@ _end:
return pBlock != NULL;
}
SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
int32_t getAggregateResultNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SAggOperatorInfo* pAggInfo = pOperator->info;
SOptrBasicInfo* pInfo = &pAggInfo->binfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -291,10 +292,18 @@ SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return (rows == 0) ? NULL : pInfo->pRes;
(*ppRes) = (rows == 0) ? NULL : pInfo->pRes;
return code;
}
static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = getAggregateResultNext(pOperator, &pRes);
return pRes;
}
int32_t doAggregateImpl(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx) {

88
source/libs/executor/src/cachescanoperator.c
View File

@ -144,7 +144,7 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
for (int i = 0; i < TARRAY_SIZE(pInfo->matchInfo.pList); ++i) {
SColMatchItem* pColInfo = taosArrayGet(pInfo->matchInfo.pList, i);
void* tmp = taosArrayPush(pCidList, &pColInfo->colId);
void* tmp = taosArrayPush(pCidList, &pColInfo->colId);
QUERY_CHECK_NULL(tmp, code, lino, _error, TSDB_CODE_OUT_OF_MEMORY);
if (pInfo->pFuncTypeList != NULL && taosArrayGetSize(pInfo->pFuncTypeList) > i) {
pColInfo->funcType = *(int32_t*)taosArrayGet(pInfo->pFuncTypeList, i);
@ -219,9 +219,12 @@ _error:
return NULL;
}
SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
int32_t doScanCacheNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SCacheRowsScanInfo* pInfo = pOperator->info;
@ -234,7 +237,8 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
int32_t size = tableListGetSize(pTableList);
if (size == 0) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
blockDataCleanup(pInfo->pRes);
@ -249,11 +253,9 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
blockDataCleanup(pBufRes);
taosArrayClear(pInfo->pUidList);
int32_t code =
code =
pReaderFn->retrieveRows(pInfo->pLastrowReader, pBufRes, pInfo->pSlotIds, pInfo->pDstSlotIds, pInfo->pUidList);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
QUERY_CHECK_CODE(code, lino, _end);
// check for tag values
int32_t resultRows = pBufRes->info.rows;
@ -277,12 +279,9 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
colDataSetNULL(pDst, 0);
} else {
if (pSrc->pData) {
char* p = colDataGetData(pSrc, pInfo->indexOfBufferedRes);
int32_t code = colDataSetVal(pDst, 0, p, false);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pTaskInfo->env, code);
}
char* p = colDataGetData(pSrc, pInfo->indexOfBufferedRes);
code = colDataSetVal(pDst, 0, p, false);
QUERY_CHECK_CODE(code, lino, _end);
}
}
}
@ -292,19 +291,22 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
pRes->info.scanFlag = MAIN_SCAN;
SExprSupp* pSup = &pInfo->pseudoExprSup;
int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes,
pRes->info.rows, pTaskInfo, NULL);
code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes, pRes->info.rows,
pTaskInfo, NULL);
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
return NULL;
(*ppRes) = NULL;
return code;
}
pRes->info.id.groupId = tableListGetTableGroupId(pTableList, pRes->info.id.uid);
pInfo->indexOfBufferedRes += 1;
return pRes;
(*ppRes) = pRes;
return code;
} else {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
} else {
size_t totalGroups = tableListGetOutputGroups(pTableList);
@ -317,37 +319,30 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
STableKeyInfo* pList = NULL;
int32_t num = 0;
int32_t code = tableListGetGroupList(pTableList, pInfo->currentGroupIndex, &pList, &num);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
code = tableListGetGroupList(pTableList, pInfo->currentGroupIndex, &pList, &num);
QUERY_CHECK_CODE(code, lino, _end);
if (NULL == pInfo->pLastrowReader) {
code = pReaderFn->openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
taosArrayGetSize(pInfo->matchInfo.pList), pInfo->pCidList, pInfo->pSlotIds, suid,
&pInfo->pLastrowReader, pTaskInfo->id.str, pInfo->pFuncTypeList, &pInfo->pkCol,
pInfo->numOfPks);
int32_t tmpRes = pReaderFn->openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
taosArrayGetSize(pInfo->matchInfo.pList), pInfo->pCidList,
pInfo->pSlotIds, suid, &pInfo->pLastrowReader, pTaskInfo->id.str,
pInfo->pFuncTypeList, &pInfo->pkCol, pInfo->numOfPks);
if (code != TSDB_CODE_SUCCESS) {
if (tmpRes != TSDB_CODE_SUCCESS) {
pInfo->currentGroupIndex += 1;
taosArrayClear(pInfo->pUidList);
continue;
}
} else {
code = pReaderFn->reuseReader(pInfo->pLastrowReader, pList, num);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pTaskInfo->env, code);
}
QUERY_CHECK_CODE(code, lino, _end);
}
taosArrayClear(pInfo->pUidList);
code = pReaderFn->retrieveRows(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pInfo->pDstSlotIds,
pInfo->pUidList);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
QUERY_CHECK_CODE(code, lino, _end);
pInfo->currentGroupIndex += 1;
@ -365,13 +360,15 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
pInfo->pRes->info.rows, pTaskInfo, NULL);
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
return NULL;
(*ppRes) = NULL;
return code;
}
}
}
// pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
} else {
// pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
}
@ -380,8 +377,23 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
pReaderFn->closeReader(pInfo->pLastrowReader);
pInfo->pLastrowReader = NULL;
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return code;
}
static SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doScanCacheNext(pOperator, &pRes);
return pRes;
}
void destroyCacheScanOperator(void* param) {

27
source/libs/executor/src/countwindowoperator.c
View File

@ -145,8 +145,8 @@ void doCountWindowAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
static void buildCountResult(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SExecTaskInfo* pTaskInfo,
SFilterInfo* pFilterInfo, SSDataBlock* pBlock) {
SResultRow* pResultRow = NULL;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
for (int32_t i = 0; i < taosArrayGetSize(pCountSup->pWinStates); i++) {
SCountWindowResult* pBuff = NULL;
code = setCountWindowOutputBuff(pExprSup, pCountSup, &pResultRow, &pBuff);
@ -171,7 +171,7 @@ _end:
}
}
static SSDataBlock* countWindowAggregate(SOperatorInfo* pOperator) {
static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SCountWindowOperatorInfo* pInfo = pOperator->info;
@ -198,11 +198,9 @@ static SSDataBlock* countWindowAggregate(SOperatorInfo* pOperator) {
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
}
code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
QUERY_CHECK_CODE(code, lino, _end);
}
if (pInfo->groupId == 0) {
@ -214,7 +212,8 @@ static SSDataBlock* countWindowAggregate(SOperatorInfo* pOperator) {
doCountWindowAggImpl(pOperator, pBlock);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
return pRes;
(*ppRes) = pRes;
return code;
}
}
@ -223,9 +222,17 @@ static SSDataBlock* countWindowAggregate(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return pRes->info.rows == 0 ? NULL : pRes;
(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;
return code;
}
static SSDataBlock* countWindowAggregate(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = countWindowAggregateNext(pOperator, &pRes);
return pRes;
}
SOperatorInfo* createCountwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,

23
source/libs/executor/src/eventwindowoperator.c
View File

@ -168,7 +168,7 @@ void destroyEWindowOperatorInfo(void* param) {
taosMemoryFreeClear(param);
}
static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SEventWindowOperatorInfo* pInfo = pOperator->info;
@ -197,11 +197,9 @@ static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
}
code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
QUERY_CHECK_CODE(code, lino, _end);
}
code = eventWindowAggImpl(pOperator, pInfo, pBlock);
@ -211,16 +209,25 @@ static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
return pRes;
(*ppRes) = pRes;
return code;
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return pRes->info.rows == 0 ? NULL : pRes;
(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;
return code;
}
static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = eventWindowAggregateNext(pOperator, &pRes);
return pRes;
}
static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,

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

@ -230,30 +230,30 @@ static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
}
}
static SSDataBlock* loadRemoteData(SOperatorInfo* pOperator) {
static int32_t loadRemoteDataNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExchangeInfo* pExchangeInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
}
code = pOperator->fpSet._openFn(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
while (1) {
SSDataBlock* pBlock = doLoadRemoteDataImpl(pOperator);
if (pBlock == NULL) {
return NULL;
(*ppRes) = NULL;
return code;
}
pTaskInfo->code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(pTaskInfo->code));
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
}
code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
QUERY_CHECK_CODE(code, lino, _end);
if (blockDataGetNumOfRows(pBlock) == 0) {
continue;
}
@ -266,15 +266,33 @@ static SSDataBlock* loadRemoteData(SOperatorInfo* pOperator) {
} else if (status == PROJECT_RETRIEVE_DONE) {
if (pBlock->info.rows == 0) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
} else {
return pBlock;
(*ppRes) = pBlock;
return code;
}
}
} else {
return pBlock;
(*ppRes) = pBlock;
return code;
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
(*ppRes) = NULL;
return code;
}
static SSDataBlock* loadRemoteData(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = loadRemoteDataNext(pOperator, &pRes);
return pRes;
}
static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const char* id) {

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

@ -316,12 +316,14 @@ _end:
return NULL;
}
static SSDataBlock* doFill(SOperatorInfo* pOperator) {
static int32_t doFillNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
SFillOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SSDataBlock* fillResult = NULL;
@ -332,9 +334,10 @@ static SSDataBlock* doFill(SOperatorInfo* pOperator) {
break;
}
int32_t code = doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo);
code = doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo);
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 (fillResult->info.rows > 0) {
@ -346,7 +349,14 @@ static SSDataBlock* doFill(SOperatorInfo* pOperator) {
pOperator->resultInfo.totalRows += fillResult->info.rows;
}
return fillResult;
(*ppRes) = fillResult;
return code;
}
static SSDataBlock* doFill(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doFillNext(pOperator, &pRes);
return pRes;
}
void destroyFillOperatorInfo(void* param) {
@ -374,8 +384,9 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
// STimeWindow w = {0};
// getInitialStartTimeWindow(pInterval, startKey, &w, order == TSDB_ORDER_ASC);
pInfo->pFillInfo = taosCreateFillInfo(startKey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo,
pInfo->primaryTsCol, order, id, pTaskInfo);
pInfo->pFillInfo = NULL;
taosCreateFillInfo(startKey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo,
pInfo->primaryTsCol, order, id, pTaskInfo, &pInfo->pFillInfo);
if (order == TSDB_ORDER_ASC) {
pInfo->win.skey = win.skey;

83
source/libs/executor/src/groupoperator.c
View File

@ -460,9 +460,10 @@ _end:
return (pRes->info.rows == 0) ? NULL : pRes;
}
static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
static int32_t hashGroupbyAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
@ -471,7 +472,8 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
SGroupbyOperatorInfo* pInfo = pOperator->info;
if (pOperator->status == OP_RES_TO_RETURN) {
return buildGroupResultDataBlockByHash(pOperator);
(*ppRes) = buildGroupResultDataBlockByHash(pOperator);
return code;
}
SGroupResInfo* pGroupResInfo = &pInfo->groupResInfo;
@ -523,7 +525,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return buildGroupResultDataBlockByHash(pOperator);
(*ppRes) = buildGroupResultDataBlockByHash(pOperator);
return code;
}
static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = hashGroupbyAggregateNext(pOperator, &pRes);
return pRes;
}
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo) {
@ -978,9 +987,10 @@ _end:
return pInfo->binfo.pRes;
}
static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
static int32_t hashPartitionNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
@ -990,7 +1000,8 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
SSDataBlock* pRes = pInfo->binfo.pRes;
if (pOperator->status == OP_RES_TO_RETURN) {
return buildPartitionResult(pOperator);
(*ppRes) = buildPartitionResult(pOperator);
return code;
}
int64_t st = taosGetTimestampUs();
@ -1005,21 +1016,21 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
}
code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
pInfo->scalarSup.numOfExprs, NULL);
QUERY_CHECK_CODE(code, lino, _end);
}
terrno = TSDB_CODE_SUCCESS;
doHashPartition(pOperator, pBlock);
if (terrno != TSDB_CODE_SUCCESS) { // group by json error
T_LONG_JMP(pTaskInfo->env, terrno);
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
SArray* groupArray = taosArrayInit(taosHashGetSize(pInfo->pGroupSet), sizeof(SDataGroupInfo));
QUERY_CHECK_NULL(groupArray, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
void* pGroupIter = taosHashIterate(pInfo->pGroupSet, NULL);
while (pGroupIter != NULL) {
@ -1043,10 +1054,18 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return buildPartitionResult(pOperator);
(*ppRes) = buildPartitionResult(pOperator);
return code;
}
static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = hashPartitionNext(pOperator, &pRes);
return pRes;
}
static void destroyPartitionOperatorInfo(void* param) {
@ -1413,26 +1432,29 @@ _end:
}
}
static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
static int32_t doStreamHashPartitionNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStreamPartitionOperatorInfo* pInfo = pOperator->info;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
if (hasRemainTbName(pInfo)) {
code = buildStreamCreateTableResult(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
if (pInfo->pCreateTbRes && pInfo->pCreateTbRes->info.rows > 0) {
return pInfo->pCreateTbRes;
(*ppRes) = pInfo->pCreateTbRes;
return code;
}
}
if (hasRemainPartion(pInfo)) {
return buildStreamPartitionResult(pOperator);
(*ppRes) = buildStreamPartitionResult(pOperator);
return code;
}
int64_t st = taosGetTimestampUs();
@ -1442,7 +1464,8 @@ static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);
if (pBlock == NULL) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "recv", GET_TASKID(pTaskInfo));
switch (pBlock->info.type) {
@ -1457,13 +1480,15 @@ static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;
printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pDelRes;
(*ppRes) = pInfo->pDelRes;
return code;
} break;
case STREAM_CREATE_CHILD_TABLE:
case STREAM_RETRIEVE:
case STREAM_CHECKPOINT:
case STREAM_GET_ALL: {
return pBlock;
(*ppRes) = pBlock;
return code;
}
default:
ASSERTS(0, "invalid SSDataBlock type");
@ -1485,15 +1510,25 @@ static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
code = buildStreamCreateTableResult(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
if (pInfo->pCreateTbRes && pInfo->pCreateTbRes->info.rows > 0) {
return pInfo->pCreateTbRes;
(*ppRes) = pInfo->pCreateTbRes;
return code;
}
return buildStreamPartitionResult(pOperator);
(*ppRes) = buildStreamPartitionResult(pOperator);
return code;
_end:
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStreamHashPartitionNext(pOperator, &pRes);
return pRes;
}
static void destroyStreamPartitionOperatorInfo(void* param) {

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

@ -842,23 +842,24 @@ static SSDataBlock* getBlockForEmptyTable(SOperatorInfo* pOperator, const STable
return pBlock;
}
static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
static int32_t doTableScanImplNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
STableScanInfo* pTableScanInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStorageAPI* pAPI = &pTaskInfo->storageAPI;
SSDataBlock* pBlock = pTableScanInfo->pResBlock;
bool hasNext = false;
int32_t code = TSDB_CODE_SUCCESS;
pBlock->info.dataLoad = false;
int64_t st = taosGetTimestampUs();
while (true) {
code = pAPI->tsdReader.tsdNextDataBlock(pTableScanInfo->base.dataReader, &hasNext);
if (code) {
if (code != TSDB_CODE_SUCCESS) {
pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);
T_LONG_JMP(pTaskInfo->env, code);
QUERY_CHECK_CODE(code, lino, _end);
}
if (!hasNext) {
@ -887,9 +888,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
uint32_t status = 0;
code = loadDataBlock(pOperator, &pTableScanInfo->base, pBlock, &status);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
QUERY_CHECK_CODE(code, lino, _end);
if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) {
break;
@ -905,9 +904,24 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
pOperator->cost.totalCost = pTableScanInfo->base.readRecorder.elapsedTime;
pBlock->info.scanFlag = pTableScanInfo->base.scanFlag;
return pBlock;
(*ppRes) = pBlock;
return code;
}
return NULL;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTableScanImplNext(pOperator, &pRes);
return pRes;
}
static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {
@ -1177,7 +1191,7 @@ _end:
return result;
}
static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
static int32_t doTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
STableScanInfo* pInfo = pOperator->info;
@ -1189,10 +1203,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
code = createTableListInfoFromParam(pOperator);
freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);
pOperator->pOperatorGetParam = NULL;
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
QUERY_CHECK_CODE(code, lino, _end);
if (pOperator->status == OP_EXEC_DONE) {
pInfo->currentGroupId = -1;
pOperator->status = OP_OPENED;
@ -1200,7 +1212,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
while (true) {
result = startNextGroupScan(pOperator);
if (result || pOperator->status == OP_EXEC_DONE) {
return result;
(*ppRes) = result;
return code;
}
}
}
@ -1215,7 +1228,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
while (1) {
SSDataBlock* result = doGroupedTableScan(pOperator);
if (result || (pOperator->status == OP_EXEC_DONE) || isTaskKilled(pTaskInfo)) {
return result;
(*ppRes) = result;
return code;
}
// if no data, switch to next table and continue scan
@ -1227,7 +1241,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
if (pInfo->currentTable >= numOfTables) {
qDebug("all table checked in table list, total:%d, return NULL, %s", numOfTables, GET_TASKID(pTaskInfo));
taosRUnLockLatch(&pTaskInfo->lock);
return NULL;
(*ppRes) = NULL;
return code;
}
tInfo = *(STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->currentTable);
@ -1243,7 +1258,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
pInfo->scanTimes = 0;
}
} else { // scan table group by group sequentially
return groupSeqTableScan(pOperator);
(*ppRes) = groupSeqTableScan(pOperator);
return code;
}
_end:
@ -1252,7 +1268,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTableScanNext(pOperator, &pRes);
return pRes;
}
static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
@ -2648,7 +2671,7 @@ static void processPrimaryKey(SSDataBlock* pBlock, bool hasPrimaryKey, STqOffset
tqOffsetResetToData(offset, pBlock->info.id.uid, pBlock->info.window.ekey, val);
}
static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
static int32_t doQueueScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -2660,7 +2683,8 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
qDebug("start to exec queue scan, %s", id);
if (isTaskKilled(pTaskInfo)) {
return NULL;
(*ppRes) = NULL;
return code;
}
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
@ -2672,7 +2696,8 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
processPrimaryKey(pResult, hasPrimaryKey, &pTaskInfo->streamInfo.currentOffset);
qDebug("tmqsnap doQueueScan get data uid:%" PRId64 "", pResult->info.id.uid);
if (pResult->info.rows > 0) {
return pResult;
(*ppRes) = pResult;
return code;
}
} else {
break;
@ -2686,7 +2711,8 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
int64_t validVer = pTaskInfo->streamInfo.snapshotVer + 1;
qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", validVer);
if (pAPI->tqReaderFn.tqReaderSeek(pInfo->tqReader, validVer, pTaskInfo->id.str) < 0) {
return NULL;
(*ppRes) = NULL;
return code;
}
tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, validVer);
@ -2716,23 +2742,34 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
qDebug("doQueueScan after filter get data from log %" PRId64 " rows, version:%" PRId64, pInfo->pRes->info.rows,
pTaskInfo->streamInfo.currentOffset.version);
if (pInfo->pRes->info.rows > 0) {
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
} else {
qDebug("doQueueScan get none from log, return, version:%" PRId64, pTaskInfo->streamInfo.currentOffset.version);
return NULL;
(*ppRes) = NULL;
return code;
}
}
} else {
qError("unexpected streamInfo prepare type: %d", pTaskInfo->streamInfo.currentOffset.type);
return NULL;
(*ppRes) = NULL;
return code;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doQueueScanNext(pOperator, &pRes);
return pRes;
}
static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, SStreamScanInfo* pInfo) {
@ -2923,7 +2960,7 @@ static bool isStreamWindow(SStreamScanInfo* pInfo) {
return isIntervalWindow(pInfo) || isSessionWindow(pInfo) || isStateWindow(pInfo) || isCountWindow(pInfo);
}
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
static int32_t doStreamScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
// NOTE: this operator does never check if current status is done or not
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
@ -2973,7 +3010,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
if (pStreamInfo->recoverStep == STREAM_RECOVER_STEP__SCAN1) {
if (isTaskKilled(pTaskInfo)) {
qInfo("===stream===stream scan is killed. task id:%s, code %s", id, tstrerror(pTaskInfo->code));
return NULL;
(*ppRes) = NULL;
return code;
}
switch (pInfo->scanMode) {
@ -2981,7 +3019,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
printSpecDataBlock(pInfo->pRecoverRes, getStreamOpName(pOperator->operatorType), "recover",
GET_TASKID(pTaskInfo));
return pInfo->pRecoverRes;
(*ppRes) = pInfo->pRecoverRes;
return code;
} break;
default:
break;
@ -3002,13 +3041,15 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pInfo->scanMode = STREAM_SCAN_FROM_RES;
printSpecDataBlock(pInfo->pCreateTbRes, getStreamOpName(pOperator->operatorType), "recover",
GET_TASKID(pTaskInfo));
return pInfo->pCreateTbRes;
(*ppRes) = pInfo->pCreateTbRes;
return code;
}
qDebug("stream recover scan get block, rows %" PRId64, pInfo->pRecoverRes->info.rows);
printSpecDataBlock(pInfo->pRecoverRes, getStreamOpName(pOperator->operatorType), "recover",
GET_TASKID(pTaskInfo));
return pInfo->pRecoverRes;
(*ppRes) = pInfo->pRecoverRes;
return code;
}
pStreamInfo->recoverStep = STREAM_RECOVER_STEP__NONE;
STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
@ -3020,7 +3061,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pTSInfo->base.cond.endVersion = -1;
pStreamInfo->recoverScanFinished = true;
return NULL;
(*ppRes) = NULL;
return code;
}
size_t total = taosArrayGetSize(pInfo->pBlockLists);
@ -3029,7 +3071,8 @@ FETCH_NEXT_BLOCK:
if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
if (pInfo->validBlockIndex >= total) {
doClearBufferedBlocks(pInfo);
return NULL;
(*ppRes) = NULL;
return code;
}
int32_t current = pInfo->validBlockIndex++;
@ -3060,7 +3103,8 @@ FETCH_NEXT_BLOCK:
case STREAM_GET_ALL:
printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
setStreamOperatorState(&pInfo->basic, pBlock->info.type);
return pBlock;
(*ppRes) = pBlock;
return code;
case STREAM_RETRIEVE: {
pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
@ -3107,7 +3151,8 @@ FETCH_NEXT_BLOCK:
printSpecDataBlock(pInfo->pDeleteDataRes, getStreamOpName(pOperator->operatorType), "delete result",
GET_TASKID(pTaskInfo));
setStreamOperatorState(&pInfo->basic, pInfo->pDeleteDataRes->info.type);
return pInfo->pDeleteDataRes;
(*ppRes) = pInfo->pDeleteDataRes;
return code;
} else {
goto FETCH_NEXT_BLOCK;
}
@ -3128,7 +3173,8 @@ FETCH_NEXT_BLOCK:
printSpecDataBlock(pInfo->pDeleteDataRes, getStreamOpName(pOperator->operatorType), "delete result",
GET_TASKID(pTaskInfo));
setStreamOperatorState(&pInfo->basic, pInfo->pDeleteDataRes->info.type);
return pInfo->pDeleteDataRes;
(*ppRes) = pInfo->pDeleteDataRes;
return code;
} else {
goto FETCH_NEXT_BLOCK;
}
@ -3142,7 +3188,8 @@ FETCH_NEXT_BLOCK:
}
printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
setStreamOperatorState(&pInfo->basic, pBlock->info.type);
return pBlock;
(*ppRes) = pBlock;
return code;
} else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) {
qDebug("stream scan mode:%d, %s", pInfo->scanMode, id);
switch (pInfo->scanMode) {
@ -3158,7 +3205,8 @@ FETCH_NEXT_BLOCK:
QUERY_CHECK_CODE(code, lino, _end);
if (pInfo->pRes->info.rows > 0) {
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
} break;
case STREAM_SCAN_FROM_DELETE_DATA: {
@ -3169,14 +3217,16 @@ FETCH_NEXT_BLOCK:
code = copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
QUERY_CHECK_CODE(code, lino, _end);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
return pInfo->pDeleteDataRes;
(*ppRes) = pInfo->pDeleteDataRes;
return code;
} break;
case STREAM_SCAN_FROM_UPDATERES: {
code = generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes, STREAM_CLEAR);
QUERY_CHECK_CODE(code, lino, _end);
prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex, NULL);
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
return pInfo->pUpdateRes;
(*ppRes) = pInfo->pUpdateRes;
return code;
} break;
case STREAM_SCAN_FROM_DATAREADER_RANGE:
case STREAM_SCAN_FROM_DATAREADER_RETRIEVE: {
@ -3193,7 +3243,8 @@ FETCH_NEXT_BLOCK:
printSpecDataBlock(pSDB, getStreamOpName(pOperator->operatorType), "update", GET_TASKID(pTaskInfo));
code = calBlockTbName(pInfo, pSDB, 0);
QUERY_CHECK_CODE(code, lino, _end);
return pSDB;
(*ppRes) = pSDB;
return code;
}
blockDataCleanup(pInfo->pUpdateDataRes);
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
@ -3212,7 +3263,8 @@ FETCH_NEXT_BLOCK:
prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex, NULL);
pInfo->pUpdateRes->info.type = STREAM_DELETE_DATA;
printSpecDataBlock(pInfo->pUpdateRes, getStreamOpName(pOperator->operatorType), "rebuild", GET_TASKID(pTaskInfo));
return pInfo->pUpdateRes;
(*ppRes) = pInfo->pUpdateRes;
return code;
}
SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
@ -3226,7 +3278,8 @@ FETCH_NEXT_BLOCK:
doClearBufferedBlocks(pInfo);
qDebug("stream scan return empty, all %d submit blocks consumed, %s", totalBlocks, id);
return NULL;
(*ppRes) = NULL;
return code;
}
int32_t current = pInfo->validBlockIndex++;
@ -3264,7 +3317,8 @@ FETCH_NEXT_BLOCK:
pInfo->scanMode = STREAM_SCAN_FROM_RES;
qDebug("create table res exists, rows:%" PRId64 " return from stream scan, %s",
pInfo->pCreateTbRes->info.rows, id);
return pInfo->pCreateTbRes;
(*ppRes) = pInfo->pCreateTbRes;
return code;
}
code = doCheckUpdate(pInfo, pBlockInfo->window.ekey, pInfo->pRes);
@ -3297,7 +3351,8 @@ FETCH_NEXT_BLOCK:
qDebug("stream scan completed, and return source rows:%" PRId64 ", %s", pBlockInfo->rows, id);
if (pBlockInfo->rows > 0) {
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
if (pInfo->pUpdateDataRes->info.rows > 0) {
@ -3308,7 +3363,8 @@ FETCH_NEXT_BLOCK:
} else if (pInfo->blockType == STREAM_INPUT__CHECKPOINT) {
if (pInfo->validBlockIndex >= total) {
doClearBufferedBlocks(pInfo);
return NULL;
(*ppRes) = NULL;
return code;
}
int32_t current = pInfo->validBlockIndex++;
@ -3321,16 +3377,24 @@ FETCH_NEXT_BLOCK:
streamScanOperatorSaveCheckpoint(pInfo);
}
// printDataBlock(pInfo->pCheckpointRes, "stream scan ck", GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
(*ppRes) = pInfo->pCheckpointRes;
return code;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStreamScanNext(pOperator, &pRes);
return pRes;
}
static int32_t extractTableIdList(const STableListInfo* pTableListInfo, SArray** ppArrayRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
@ -3353,7 +3417,7 @@ _end:
return code;
}
static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
static int32_t doRawScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -3368,9 +3432,9 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
bool hasNext = false;
if (pInfo->dataReader && pInfo->sContext->withMeta != ONLY_META) {
code = pAPI->tsdReader.tsdNextDataBlock(pInfo->dataReader, &hasNext);
if (code) {
if (code != TSDB_CODE_SUCCESS) {
pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->dataReader);
T_LONG_JMP(pTaskInfo->env, code);
QUERY_CHECK_CODE(code, lino, _end);
}
}
@ -3382,15 +3446,14 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
SSDataBlock* pBlock = NULL;
code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pInfo->dataReader, &pBlock, NULL);
if (pBlock == NULL || code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
QUERY_CHECK_CODE(code, lino, _end);
if (pBlock && pBlock->info.rows > 0) {
bool hasPrimaryKey = pAPI->snapshotFn.taosXGetTablePrimaryKey(pInfo->sContext);
processPrimaryKey(pBlock, hasPrimaryKey, &pTaskInfo->streamInfo.currentOffset);
qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.id.uid);
return pBlock;
(*ppRes) = pBlock;
return code;
}
}
@ -3407,7 +3470,8 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
code = qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
QUERY_CHECK_CODE(code, lino, _end);
tDeleteSchemaWrapper(mtInfo.schema);
return NULL;
(*ppRes) = NULL;
return code;
} else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {
SSnapContext* sContext = pInfo->sContext;
for (int32_t i = 0; i < tmqRowSize; i++) {
@ -3441,22 +3505,24 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
pTaskInfo->streamInfo.btMetaRsp.batchMetaLen = taosArrayInit(4, sizeof(int32_t));
QUERY_CHECK_NULL(pTaskInfo->streamInfo.btMetaRsp.batchMetaLen, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t tempRes = TSDB_CODE_SUCCESS;
uint32_t len = 0;
tEncodeSize(tEncodeMqMetaRsp, &tmpMetaRsp, len, code);
if (TSDB_CODE_SUCCESS != code) {
tEncodeSize(tEncodeMqMetaRsp, &tmpMetaRsp, len, tempRes);
if (TSDB_CODE_SUCCESS != tempRes) {
qError("tmqsnap tEncodeMqMetaRsp error");
taosMemoryFreeClear(data);
break;
}
int32_t tLen = sizeof(SMqRspHead) + len;
void* tBuf = taosMemoryCalloc(1, tLen);
int32_t tLen = sizeof(SMqRspHead) + len;
void* tBuf = taosMemoryCalloc(1, tLen);
QUERY_CHECK_NULL(tBuf, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
void* metaBuff = POINTER_SHIFT(tBuf, sizeof(SMqRspHead));
SEncoder encoder = {0};
tEncoderInit(&encoder, metaBuff, len);
code = tEncodeMqMetaRsp(&encoder, &tmpMetaRsp);
if (code < 0) {
int32_t tempLen = tEncodeMqMetaRsp(&encoder, &tmpMetaRsp);
if (tempLen < 0) {
qError("tmqsnap tEncodeMqMetaRsp error");
tEncoderClear(&encoder);
taosMemoryFreeClear(tBuf);
@ -3471,7 +3537,8 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
QUERY_CHECK_NULL(tmp, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
}
}
return NULL;
(*ppRes) = NULL;
return code;
}
_end:
@ -3480,7 +3547,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doRawScanNext(pOperator, &pRes);
return pRes;
}
static void destroyRawScanOperatorInfo(void* param) {
@ -3570,6 +3644,7 @@ void streamScanReleaseState(SOperatorInfo* pOperator) {
return;
}
if (!pInfo->pUpdateInfo) {
qDebug("stask:%s streamScanReleaseState cancel", GET_TASKID(pOperator->pTaskInfo));
return;
}
int32_t len = 0;
@ -3602,6 +3677,10 @@ void streamScanReloadState(SOperatorInfo* pOperator) {
if (!pInfo->pState) {
return;
}
if (!pInfo->pUpdateInfo) {
qDebug("stask:%s streamScanReloadState cancel", GET_TASKID(pOperator->pTaskInfo));
return;
}
void* pBuff = NULL;
int32_t len = 0;
code = pInfo->stateStore.streamStateGetInfo(pInfo->pState, STREAM_SCAN_OP_STATE_NAME,
@ -4179,9 +4258,10 @@ _end:
return code;
}
static SSDataBlock* doTagScanFromCtbIdx(SOperatorInfo* pOperator) {
static int32_t doTagScanFromCtbIdxNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
@ -4268,12 +4348,21 @@ _end:
T_LONG_JMP(pTaskInfo->env, code);
}
pOperator->resultInfo.totalRows += pRes->info.rows;
return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
(*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;
return code;
}
static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {
static SSDataBlock* doTagScanFromCtbIdx(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTagScanFromCtbIdxNext(pOperator, &pRes);
return pRes;
}
static int32_t doTagScanFromMetaEntryNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -4287,7 +4376,8 @@ static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {
int32_t size = tableListGetSize(pInfo->pTableListInfo);
if (size == 0) {
setTaskStatus(pTaskInfo, TASK_COMPLETED);
return NULL;
(*ppRes) = NULL;
return code;
}
char str[512] = {0};
@ -4316,7 +4406,14 @@ static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {
pOperator->resultInfo.totalRows += pRes->info.rows;
return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
(*ppRes) = (pRes->info.rows == 0) ? NULL : pInfo->pRes;
return code;
}
static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTagScanFromMetaEntryNext(pOperator, &pRes);
return pRes;
}
static void destroyTagScanOperatorInfo(void* param) {
@ -4849,9 +4946,10 @@ static void stopSubTablesTableMergeScan(STableMergeScanInfo* pInfo) {
}
}
SSDataBlock* doTableMergeScanParaSubTables(SOperatorInfo* pOperator) {
int32_t doTableMergeScanParaSubTablesNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -4871,7 +4969,8 @@ SSDataBlock* doTableMergeScanParaSubTables(SOperatorInfo* pOperator) {
if (tableListSize == 0) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
pInfo->tableStartIndex = 0;
pInfo->groupId = ((STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex))->groupId;
@ -4919,7 +5018,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return pBlock;
(*ppRes) = pBlock;
return code;
}
static SSDataBlock* doTableMergeScanParaSubTables(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTableMergeScanParaSubTablesNext(pOperator, &pRes);
return pRes;
}
static void tableMergeScanDoSkipTable(uint64_t uid, void* pTableMergeOpInfo) {
@ -5312,18 +5418,19 @@ SSDataBlock* getSortedTableMergeScanBlockData(SSortHandle* pHandle, SSDataBlock*
return (pResBlock->info.rows > 0) ? pResBlock : NULL;
}
SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
int32_t doTableMergeScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableMergeScanInfo* pInfo = pOperator->info;
int32_t code = pOperator->fpSet._openFn(pOperator);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
code = pOperator->fpSet._openFn(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
int64_t st = taosGetTimestampUs();
@ -5333,7 +5440,8 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
if (tableListSize == 0) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
pInfo->tableStartIndex = 0;
pInfo->groupId = ((STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->tableStartIndex))->groupId;
@ -5361,9 +5469,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
if (pInfo->bNewFilesetEvent) {
stopDurationForGroupTableMergeScan(pOperator);
code = startDurationForGroupTableMergeScan(pOperator);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
QUERY_CHECK_CODE(code, lino, _end);
} else {
// Data of this group are all dumped, let's try the next group
stopGroupTableMergeScan(pOperator);
@ -5382,7 +5488,20 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
pOperator->cost.totalCost += (taosGetTimestampUs() - st) / 1000.0;
return pBlock;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
(*ppRes) = pBlock;
return code;
}
static SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTableMergeScanNext(pOperator, &pRes);
return pRes;
}
void destroyTableMergeScanOperatorInfo(void* param) {
@ -5841,7 +5960,8 @@ _end:
}
}
static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) {
static int32_t doTableCountScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableCountScanOperatorInfo* pInfo = pOperator->info;
STableCountScanSupp* pSupp = &pInfo->supp;
@ -5849,13 +5969,22 @@ static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) {
blockDataCleanup(pRes);
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
if (pInfo->readHandle.mnd != NULL) {
return buildSysDbTableCount(pOperator, pInfo);
(*ppRes) = buildSysDbTableCount(pOperator, pInfo);
return code;
}
return buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);
(*ppRes) = buildVnodeDbTableCount(pOperator, pInfo, pSupp, pRes);
return code;
}
static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTableCountScanNext(pOperator, &pRes);
return pRes;
}
static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo,

76
source/libs/executor/src/streamcountwindowoperator.c
View File

@ -357,12 +357,13 @@ static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
destroySBuffInfo(pAggSup, &buffInfo);
}
static SSDataBlock* buildCountResult(SOperatorInfo* pOperator) {
static int32_t buildCountResult(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
SStreamCountAggOperatorInfo* pInfo = pOperator->info;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
SOptrBasicInfo* pBInfo = &pInfo->binfo;
@ -370,15 +371,18 @@ static SSDataBlock* buildCountResult(SOperatorInfo* pOperator) {
doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pDelRes;
(*ppRes) = pInfo->pDelRes;
return code;
}
doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
if (pBInfo->pRes->info.rows > 0) {
printDataBlock(pBInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pBInfo->pRes;
(*ppRes) = pBInfo->pRes;
return code;
}
return NULL;
(*ppRes) = NULL;
return code;
}
int32_t doStreamCountEncodeOpState(void** buf, int32_t len, SOperatorInfo* pOperator, bool isParent) {
@ -425,6 +429,7 @@ int32_t doStreamCountDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamCountAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (!pInfo) {
code = TSDB_CODE_FAILED;
QUERY_CHECK_CODE(code, lino, _end);
@ -465,7 +470,7 @@ int32_t doStreamCountDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -475,6 +480,7 @@ void doStreamCountSaveCheckpoint(SOperatorInfo* pOperator) {
int32_t lino = 0;
void* pBuf = NULL;
SStreamCountAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (needSaveStreamOperatorInfo(&pInfo->basic)) {
int32_t len = doStreamCountEncodeOpState(NULL, 0, pOperator, true);
pBuf = taosMemoryCalloc(1, len);
@ -492,7 +498,7 @@ void doStreamCountSaveCheckpoint(SOperatorInfo* pOperator) {
_end:
taosMemoryFreeClear(pBuf);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -604,7 +610,7 @@ _end:
return code;
}
static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
static int32_t doStreamCountAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExprSupp* pSup = &pOperator->exprSupp;
@ -614,11 +620,15 @@ static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
qDebug("stask:%s %s status: %d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status);
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
} else if (pOperator->status == OP_RES_TO_RETURN) {
SSDataBlock* opRes = buildCountResult(pOperator);
SSDataBlock* opRes = NULL;
code = buildCountResult(pOperator, &opRes);
QUERY_CHECK_CODE(code, lino, _end);
if (opRes) {
return opRes;
(*ppRes) = opRes;
return code;
}
if (pInfo->recvGetAll) {
@ -629,11 +639,13 @@ static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
(*ppRes) = pInfo->pCheckpointRes;
return code;
}
setStreamOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
SOperatorInfo* downstream = pOperator->pDownstream[0];
@ -667,7 +679,8 @@ static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
continue;
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
return pBlock;
(*ppRes) = pBlock;
return code;
} else if (pBlock->info.type == STREAM_CHECKPOINT) {
pAggSup->stateStore.streamStateCommit(pAggSup->pState);
doStreamCountSaveCheckpoint(pOperator);
@ -710,29 +723,39 @@ static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
}
SSDataBlock* opRes = buildCountResult(pOperator);
SSDataBlock* opRes = NULL;
code = buildCountResult(pOperator, &opRes);
QUERY_CHECK_CODE(code, lino, _end);
if (opRes) {
return opRes;
(*ppRes) = opRes;
return code;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
setStreamOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStreamCountAggNext(pOperator, &pRes);
return pRes;
}
void streamCountReleaseState(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
int32_t resSize = sizeof(TSKEY);
char* pBuff = taosMemoryCalloc(1, resSize);
if (pBuff) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
QUERY_CHECK_NULL(pBuff, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
memcpy(pBuff, &pInfo->twAggSup.maxTs, sizeof(TSKEY));
qDebug("===stream=== count window operator relase state. ");
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_STATE_NAME,
@ -745,7 +768,8 @@ void streamCountReleaseState(SOperatorInfo* pOperator) {
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code;
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -753,6 +777,7 @@ void streamCountReloadState(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamCountAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
int32_t size = 0;
void* pBuf = NULL;
@ -773,7 +798,8 @@ void streamCountReloadState(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code;
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -882,6 +908,6 @@ _error:
taosMemoryFreeClear(pOperator);
pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
return NULL;
}

74
source/libs/executor/src/streameventwindowoperator.c
View File

@ -285,7 +285,7 @@ static int32_t compactEventWindow(SOperatorInfo* pOperator, SEventWindowInfo* pC
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -382,8 +382,8 @@ static void doStreamEventAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
&curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL);
QUERY_CHECK_CODE(code, lino, _end);
code = tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey));
QUERY_CHECK_CODE(code, lino, _end);
int32_t tmpRes = tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey));
qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
doDeleteEventWindow(pAggSup, pSeUpdated, &curWin.winInfo.sessionWin);
if (pInfo->destHasPrimaryKey && curWin.winInfo.isOutput && IS_NORMAL_EVENT_OP(pOperator) &&
@ -444,7 +444,7 @@ _end:
colDataDestroy(pColEnd);
taosMemoryFree(pColEnd);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -490,6 +490,7 @@ int32_t doStreamEventDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (!pInfo) {
code = TSDB_CODE_FAILED;
QUERY_CHECK_CODE(code, lino, _end);
@ -532,7 +533,7 @@ int32_t doStreamEventDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -555,7 +556,8 @@ void doStreamEventSaveCheckpoint(SOperatorInfo* pOperator) {
}
}
static SSDataBlock* buildEventResult(SOperatorInfo* pOperator) {
static int32_t buildEventResult(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SOptrBasicInfo* pBInfo = &pInfo->binfo;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -563,20 +565,24 @@ static SSDataBlock* buildEventResult(SOperatorInfo* pOperator) {
doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pDelRes;
(*ppRes) = pInfo->pDelRes;
return code;
}
doBuildSessionResult(pOperator, pInfo->streamAggSup.pState, &pInfo->groupResInfo, pBInfo->pRes);
if (pBInfo->pRes->info.rows > 0) {
printDataBlock(pBInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pBInfo->pRes;
(*ppRes) = pBInfo->pRes;
return code;
}
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
static int32_t doStreamEventAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
@ -585,11 +591,14 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SOptrBasicInfo* pBInfo = &pInfo->binfo;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
qDebug("===stream=== stream event agg");
qDebug("===stream=== stream event agg. history task:%d, taskId:%s", pInfo->isHistoryOp, GET_TASKID(pTaskInfo));
if (pOperator->status == OP_RES_TO_RETURN) {
SSDataBlock* resBlock = buildEventResult(pOperator);
SSDataBlock* resBlock = NULL;
code = buildEventResult(pOperator, &resBlock);
QUERY_CHECK_CODE(code, lino, _end);
if (resBlock != NULL) {
return resBlock;
(*ppRes) = resBlock;
return code;
}
if (pInfo->recvGetAll) {
@ -600,11 +609,13 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
(*ppRes) = pInfo->pCheckpointRes;
return code;
}
setStreamOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
SOperatorInfo* downstream = pOperator->pDownstream[0];
@ -636,7 +647,8 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
continue;
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
return pBlock;
(*ppRes) = pBlock;
return code;
} else if (pBlock->info.type == STREAM_CHECKPOINT) {
pInfo->streamAggSup.stateStore.streamStateCommit(pInfo->streamAggSup.pState);
doStreamEventSaveCheckpoint(pOperator);
@ -680,6 +692,10 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
code = copyUpdateResult(&pInfo->pAllUpdated, pHisWins, sessionKeyCompareAsc);
QUERY_CHECK_CODE(code, lino, _end);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pInfo->pAllUpdated = tSimpleHashInit(64, hashFn);
QUERY_CHECK_NULL(pInfo->pAllUpdated, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
code = getMaxTsWins(pHisWins, pInfo->historyWins);
QUERY_CHECK_CODE(code, lino, _end);
@ -695,17 +711,28 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
code = blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _end);
SSDataBlock* resBlock = buildEventResult(pOperator);
SSDataBlock* resBlock = NULL;
code = buildEventResult(pOperator, &resBlock);
QUERY_CHECK_CODE(code, lino, _end);
if (resBlock != NULL) {
return resBlock;
(*ppRes) = resBlock;
return code;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
setStreamOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStreamEventAggNext(pOperator, &pRes);
return pRes;
}
void streamEventReleaseState(SOperatorInfo* pOperator) {
@ -733,6 +760,7 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
int32_t lino = 0;
SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
resetWinRange(&pAggSup->winRange);
SSessionKey seKey = {.win.skey = INT64_MIN, .win.ekey = INT64_MIN, .groupId = 0};
@ -816,7 +844,7 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -937,6 +965,6 @@ _error:
destroyStreamEventOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator);
pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
return NULL;
}

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

@ -563,8 +563,7 @@ static void doStreamFillLinear(SStreamFillSupporter* pFillSup, SStreamFillInfo*
code = checkResult(pFillSup, pFillInfo->current, groupId, &ckRes);
QUERY_CHECK_CODE(code, lino, _end);
if ((pFillSup->hasDelete && !ckRes) ||
!inWinRange(&pFillSup->winRange, &st)) {
if ((pFillSup->hasDelete && !ckRes) || !inWinRange(&pFillSup->winRange, &st)) {
pFillInfo->current = taosTimeAdd(pFillInfo->current, pFillSup->interval.sliding, pFillSup->interval.slidingUnit,
pFillSup->interval.precision);
pFillInfo->pLinearInfo->winIndex++;
@ -743,7 +742,7 @@ static void doStreamFillImpl(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
}
@ -752,6 +751,7 @@ static int32_t buildDeleteRange(SOperatorInfo* pOp, TSKEY start, TSKEY end, uint
int32_t lino = 0;
SStorageAPI* pAPI = &pOp->pTaskInfo->storageAPI;
void* pState = pOp->pTaskInfo->streamInfo.pState;
SExecTaskInfo* pTaskInfo = pOp->pTaskInfo;
SSDataBlock* pBlock = delRes;
SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
@ -794,7 +794,7 @@ static int32_t buildDeleteRange(SOperatorInfo* pOp, TSKEY start, TSKEY end, uint
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -805,6 +805,7 @@ static int32_t buildDeleteResult(SOperatorInfo* pOperator, TSKEY startTs, TSKEY
int32_t lino = 0;
SStreamFillOperatorInfo* pInfo = pOperator->info;
SStreamFillSupporter* pFillSup = pInfo->pFillSup;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (hasPrevWindow(pFillSup)) {
TSKEY start = getNextWindowTs(pFillSup->prev.key, &pFillSup->interval);
code = buildDeleteRange(pOperator, start, endTs, groupId, delRes);
@ -820,7 +821,7 @@ static int32_t buildDeleteResult(SOperatorInfo* pOperator, TSKEY startTs, TSKEY
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -830,6 +831,7 @@ static int32_t doDeleteFillResultImpl(SOperatorInfo* pOperator, TSKEY startTs, T
int32_t lino = 0;
SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;
SStreamFillOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
getWindowFromDiscBuf(pOperator, startTs, groupId, pInfo->pFillSup);
setDeleteFillValueInfo(startTs, endTs, pInfo->pFillSup, pInfo->pFillInfo);
SWinKey key = {.ts = startTs, .groupId = groupId};
@ -852,7 +854,7 @@ static int32_t doDeleteFillResultImpl(SOperatorInfo* pOperator, TSKEY startTs, T
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -911,6 +913,7 @@ static int32_t doDeleteFillResult(SOperatorInfo* pOperator) {
SStreamFillOperatorInfo* pInfo = pOperator->info;
SStreamFillInfo* pFillInfo = pInfo->pFillInfo;
SSDataBlock* pBlock = pInfo->pSrcDelBlock;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
TSKEY* tsStarts = (TSKEY*)pStartCol->pData;
@ -967,7 +970,7 @@ static int32_t doDeleteFillResult(SOperatorInfo* pOperator) {
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
@ -985,6 +988,7 @@ static int32_t doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBl
int32_t lino = 0;
SStreamFillOperatorInfo* pInfo = pOperator->info;
SExprSupp* pSup = &pOperator->exprSupp;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
blockDataCleanup(pDstBlock);
code = blockDataEnsureCapacity(pDstBlock, pSrcBlock->info.rows);
@ -1008,19 +1012,20 @@ static int32_t doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBl
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
return code;
}
static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
static int32_t doStreamFillNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamFillOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
blockDataCleanup(pInfo->pRes);
if (hasRemainCalc(pInfo->pFillInfo) ||
@ -1028,18 +1033,21 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
doStreamFillRange(pInfo->pFillInfo, pInfo->pFillSup, pInfo->pRes);
if (pInfo->pRes->info.rows > 0) {
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
}
if (pOperator->status == OP_RES_TO_RETURN) {
doDeleteFillFinalize(pOperator);
if (pInfo->pRes->info.rows > 0) {
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
setOperatorCompleted(pOperator);
resetStreamFillInfo(pInfo);
return NULL;
(*ppRes) = NULL;
return code;
}
SSDataBlock* fillResult = NULL;
@ -1053,7 +1061,8 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
pInfo->pFillInfo->preRowKey = INT64_MIN;
if (pInfo->pRes->info.rows > 0) {
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
}
break;
}
@ -1071,7 +1080,8 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
switch (pBlock->info.type) {
case STREAM_RETRIEVE:
return pBlock;
(*ppRes) = pBlock;
return code;
case STREAM_DELETE_RESULT: {
pInfo->pSrcDelBlock = pBlock;
pInfo->srcDelRowIndex = 0;
@ -1082,7 +1092,8 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pDelRes;
(*ppRes) = pInfo->pDelRes;
return code;
}
continue;
} break;
@ -1097,7 +1108,8 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
} break;
case STREAM_CHECKPOINT:
case STREAM_CREATE_CHILD_TABLE: {
return pBlock;
(*ppRes) = pBlock;
return code;
} break;
default:
ASSERTS(false, "invalid SSDataBlock type");
@ -1121,20 +1133,30 @@ static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
if (pInfo->pRes->info.rows == 0) {
setOperatorCompleted(pOperator);
resetStreamFillInfo(pInfo);
return NULL;
(*ppRes) = NULL;
return code;
}
pOperator->resultInfo.totalRows += pInfo->pRes->info.rows;
printDataBlock(pInfo->pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pRes;
(*ppRes) = pInfo->pRes;
return code;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
setOperatorCompleted(pOperator);
resetStreamFillInfo(pInfo);
return NULL;
(*ppRes) = NULL;
return code;
}
static SSDataBlock* doStreamFill(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStreamFillNext(pOperator, &pRes);
return pRes;
}
static int32_t initResultBuf(SStreamFillSupporter* pFillSup) {
@ -1406,7 +1428,7 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
_error:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
destroyStreamFillOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator);

424
source/libs/executor/src/streamtimewindowoperator.c
View File

File diff suppressed because it is too large Load Diff

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

@ -1872,8 +1872,9 @@ static void getDBNameFromCondition(SNode* pCondition, const char* dbName) {
nodesWalkExpr(pCondition, getDBNameFromConditionWalker, (char*)dbName);
}
static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
static int32_t doSysTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
// build message and send to mnode to fetch the content of system tables.
int32_t code = TSDB_CODE_SUCCESS;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;
char dbName[TSDB_DB_NAME_LEN] = {0};
@ -1881,7 +1882,8 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
while (1) {
if (isTaskKilled(pOperator->pTaskInfo)) {
setOperatorCompleted(pOperator);
return NULL;
(*ppRes) = NULL;
return code;
}
blockDataCleanup(pInfo->pRes);
@ -1923,13 +1925,21 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
if (pBlock->info.rows == 0) {
continue;
}
return pBlock;
(*ppRes) = pBlock;
return code;
} else {
return NULL;
(*ppRes) = NULL;
return code;
}
}
}
static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doSysTableScanNext(pOperator, &pRes);
return pRes;
}
static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, const char* name,
SSDataBlock* pBlock) {
int32_t code = TSDB_CODE_SUCCESS;
@ -1974,7 +1984,11 @@ static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableSca
int32_t contLen = tSerializeSRetrieveTableReq(NULL, 0, &pInfo->req);
char* buf1 = taosMemoryCalloc(1, contLen);
(void)tSerializeSRetrieveTableReq(buf1, contLen, &pInfo->req);
int32_t tempRes = tSerializeSRetrieveTableReq(buf1, contLen, &pInfo->req);
if (tempRes < 0) {
code = terrno;
return NULL;
}
// send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
@ -2548,11 +2562,12 @@ static int32_t doGetTableRowSize(SReadHandle* pHandle, uint64_t uid, int32_t* ro
return TSDB_CODE_SUCCESS;
}
static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
static int32_t doBlockInfoScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SBlockDistInfo* pBlockScanInfo = pOperator->info;
@ -2578,7 +2593,11 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
char* p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
QUERY_CHECK_NULL(p, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
(void)tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo);
int32_t tempRes = tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo);
if (tempRes < 0) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
varDataSetLen(p, len);
code = colDataSetVal(pColInfo, 0, p, false);
@ -2602,7 +2621,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return pBlock;
(*ppRes) = pBlock;
return code;
}
static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doBlockInfoScanNext(pOperator, &pRes);
return pRes;
}
static void destroyBlockDistScanOperatorInfo(void* param) {

24
source/libs/executor/src/tfill.c
View File

@ -47,9 +47,9 @@ static void setNotFillColumn(SFillInfo* pFillInfo, SColumnInfoData* pDstColInfo,
}
SGroupKeys* pKey = taosArrayGet(p->pRowVal, colIdx);
int32_t code = doSetVal(pDstColInfo, rowIndex, pKey);
int32_t code = doSetVal(pDstColInfo, rowIndex, pKey);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
T_LONG_JMP(pFillInfo->pTaskInfo->env, code);
}
}
@ -511,20 +511,18 @@ static int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) {
return pFillInfo->numOfRows - pFillInfo->index;
}
struct SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol,
int32_t primaryTsSlotId, int32_t order, const char* id, SExecTaskInfo* pTaskInfo) {
void taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t primaryTsSlotId,
int32_t order, const char* id, SExecTaskInfo* pTaskInfo, SFillInfo** ppFillInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (fillType == TSDB_FILL_NONE) {
return NULL;
(*ppFillInfo) = NULL;
return;
}
SFillInfo* pFillInfo = taosMemoryCalloc(1, sizeof(SFillInfo));
if (pFillInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
QUERY_CHECK_NULL(pFillInfo, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
pFillInfo->order = order;
pFillInfo->srcTsSlotId = primaryTsSlotId;
@ -562,10 +560,10 @@ _end:
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(pFillInfo->next.pRowVal);
taosArrayDestroy(pFillInfo->prev.pRowVal);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
terrno = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return pFillInfo;
(*ppFillInfo) = pFillInfo;
}
void taosResetFillInfo(SFillInfo* pFillInfo, TSKEY startTimestamp) {

26
source/libs/executor/src/timesliceoperator.c
View File

@ -321,7 +321,7 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
char* v = colDataGetData(pSrc, index);
code = colDataSetVal(pDst, pResBlock->info.rows, v, false);
QUERY_CHECK_CODE(code, lino, _end);
} else if(!isSelectGroupConstValueFunc(pExprInfo)){
} else if (!isSelectGroupConstValueFunc(pExprInfo)) {
// use stored group key
SGroupKeys* pkey = pSliceInfo->pPrevGroupKey;
if (pkey->isNull == false) {
@ -331,7 +331,7 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
colDataSetNULL(pDst, rows);
}
} else {
int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, srcSlot);
if (pkey->isNull == false) {
colDataSetVal(pDst, rows, pkey->pData, false);
@ -781,10 +781,10 @@ static void saveBlockStatus(STimeSliceOperatorInfo* pSliceInfo, SSDataBlock* pBl
static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pSliceInfo, SSDataBlock* pBlock,
SExecTaskInfo* pTaskInfo, bool ignoreNull) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSDataBlock* pResBlock = pSliceInfo->pRes;
SInterval* pInterval = &pSliceInfo->interval;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSDataBlock* pResBlock = pSliceInfo->pRes;
SInterval* pInterval = &pSliceInfo->interval;
SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
SColumnInfoData* pPkCol = NULL;
@ -988,12 +988,13 @@ static void doHandleTimeslice(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
copyPrevGroupKey(&pOperator->exprSupp, pSliceInfo->pPrevGroupKey, pBlock);
}
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
static int32_t doTimesliceNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
@ -1079,7 +1080,14 @@ _finished:
T_LONG_JMP(pTaskInfo->env, code);
}
return pResBlock->info.rows == 0 ? NULL : pResBlock;
(*ppRes) = pResBlock->info.rows == 0 ? NULL : pResBlock;
return code;
}
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doTimesliceNext(pOperator, &pRes);
return pRes;
}
static int32_t extractPkColumnFromFuncs(SNodeList* pFuncs, bool* pHasPk, SColumn* pPkColumn) {

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

@ -811,9 +811,10 @@ void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOperatorInf
}
}
SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult, uint64_t groupId, SExecTaskInfo* pTaskInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult, uint64_t groupId,
SExecTaskInfo* pTaskInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SOpenWindowInfo openWin = {0};
openWin.pos.pageId = pResult->pageId;
openWin.pos.offset = pResult->offset;
@ -1053,36 +1054,28 @@ _end:
return code;
}
static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
static int32_t doStateWindowAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStateWindowOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SOptrBasicInfo* pBInfo = &pInfo->binfo;
pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
setOperatorCompleted(pOperator);
return NULL;
}
code = pOperator->fpSet._openFn(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
int32_t code = blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
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);
}
code = blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _end);
while (1) {
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
code = doFilter(pBInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
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);
}
QUERY_CHECK_CODE(code, lino, _end);
bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
if (!hasRemain) {
@ -1096,31 +1089,42 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
}
pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
(*ppRes) = (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
return code;
}
static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doStateWindowAggNext(pOperator, &pRes);
return pRes;
}
static int32_t doBuildIntervalResultNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SIntervalAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SSDataBlock* pBlock = pInfo->binfo.pRes;
pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
return NULL;
}
code = pOperator->fpSet._openFn(pOperator);
QUERY_CHECK_CODE(code, lino, _end);
while (1) {
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
int32_t code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
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);
}
code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
QUERY_CHECK_CODE(code, lino, _end);
bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
if (!hasRemain) {
@ -1136,7 +1140,20 @@ static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
size_t rows = pBlock->info.rows;
pOperator->resultInfo.totalRows += rows;
return (rows == 0) ? NULL : pBlock;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
(*ppRes) = (rows == 0) ? NULL : pBlock;
return code;
}
static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doBuildIntervalResultNext(pOperator, &pRes);
return pRes;
}
static void destroyStateWindowOperatorInfo(void* param) {
@ -1429,9 +1446,10 @@ static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperator
pRowSup->numOfRows, pBlock->info.rows, numOfOutput);
}
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
static int32_t doSessionWindowAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t code = TSDB_CODE_SUCCESS;
@ -1458,7 +1476,8 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
}
}
pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
(*ppRes) = (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
return code;
}
int64_t st = taosGetTimestampUs();
@ -1519,7 +1538,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
(*ppRes) = (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
return code;
}
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doSessionWindowAggNext(pOperator, &pRes);
return pRes;
}
// todo make this as an non-blocking operator
@ -1883,13 +1909,14 @@ _end:
}
}
static SSDataBlock* mergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
static int32_t mergeAlignedIntervalAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
int32_t code = TSDB_CODE_SUCCESS;
SMergeAlignedIntervalAggOperatorInfo* pMiaInfo = pOperator->info;
SIntervalAggOperatorInfo* iaInfo = pMiaInfo->intervalAggOperatorInfo;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SSDataBlock* pRes = iaInfo->binfo.pRes;
@ -1913,7 +1940,14 @@ static SSDataBlock* mergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
size_t rows = pRes->info.rows;
pOperator->resultInfo.totalRows += rows;
return (rows == 0) ? NULL : pRes;
(*ppRes) = (rows == 0) ? NULL : pRes;
return code;
}
static SSDataBlock* mergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = mergeAlignedIntervalAggNext(pOperator, &pRes);
return pRes;
}
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
@ -2151,7 +2185,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
}
}
static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
static int32_t doMergeIntervalAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -2161,7 +2195,8 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
SExprSupp* pExpSupp = &pOperator->exprSupp;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
(*ppRes) = NULL;
return code;
}
SSDataBlock* pRes = iaInfo->binfo.pRes;
@ -2231,7 +2266,14 @@ _end:
pTaskInfo->code = code;
T_LONG_JMP(pTaskInfo->env, code);
}
return (rows == 0) ? NULL : pRes;
(*ppRes) = (rows == 0) ? NULL : pRes;
return code;
}
static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = NULL;
int32_t code = doMergeIntervalAggNext(pOperator, &pRes);
return pRes;
}
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,

19
source/util/src/tlog.c
View File

@ -19,6 +19,7 @@
#include "tconfig.h"
#include "tglobal.h"
#include "tjson.h"
#include "tutil.h"
#define LOG_MAX_LINE_SIZE (10024)
#define LOG_MAX_LINE_BUFFER_SIZE (LOG_MAX_LINE_SIZE + 3)
@ -146,7 +147,7 @@ static int32_t taosStartLog() {
TdThreadAttr threadAttr;
taosThreadAttrInit(&threadAttr);
if (taosThreadCreate(&(tsLogObj.logHandle->asyncThread), &threadAttr, taosAsyncOutputLog, tsLogObj.logHandle) != 0) {
return -1;
return terrno;
}
taosThreadAttrDestroy(&threadAttr);
return 0;
@ -176,13 +177,13 @@ int32_t taosInitSlowLog() {
}
tsLogObj.slowHandle = taosLogBuffNew(LOG_SLOW_BUF_SIZE);
if (tsLogObj.slowHandle == NULL) return -1;
if (tsLogObj.slowHandle == NULL) return terrno;
taosUmaskFile(0);
tsLogObj.slowHandle->pFile = taosOpenFile(fullName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
if (tsLogObj.slowHandle->pFile == NULL) {
printf("\nfailed to open slow log file:%s, reason:%s\n", fullName, strerror(errno));
return -1;
return TAOS_SYSTEM_ERROR(errno);
}
return 0;
@ -209,11 +210,11 @@ int32_t taosInitLog(const char *logName, int32_t maxFiles) {
taosUpdateDaylight();
tsLogObj.logHandle = taosLogBuffNew(LOG_DEFAULT_BUF_SIZE);
if (tsLogObj.logHandle == NULL) return -1;
if (taosOpenLogFile(fullName, maxFiles) < 0) return -1;
if (tsLogObj.logHandle == NULL) return terrno;
TAOS_CHECK_RETURN(taosOpenLogFile(fullName, maxFiles));
if (taosInitSlowLog() < 0) return -1;
if (taosStartLog() < 0) return -1;
TAOS_CHECK_RETURN(taosInitSlowLog());
TAOS_CHECK_RETURN(taosStartLog());
return 0;
}
@ -484,7 +485,7 @@ static int32_t taosOpenLogFile(char *fn, int32_t maxFileNum) {
if (tsLogObj.logHandle->pFile == NULL) {
printf("\nfailed to open log file:%s, reason:%s\n", fileName, strerror(errno));
return -1;
return TAOS_SYSTEM_ERROR(errno);
}
taosLockLogFile(tsLogObj.logHandle->pFile);
@ -492,7 +493,7 @@ static int32_t taosOpenLogFile(char *fn, int32_t maxFileNum) {
int64_t filesize = 0;
if (taosFStatFile(tsLogObj.logHandle->pFile, &filesize, NULL) < 0) {
printf("\nfailed to fstat log file:%s, reason:%s\n", fileName, strerror(errno));
return -1;
return TAOS_SYSTEM_ERROR(errno);
}
tsLogObj.lines = (int32_t)(filesize / 60);

51
source/util/src/tsched.c
View File

@ -27,8 +27,10 @@ static void *taosProcessSchedQueue(void *param);
static void taosDumpSchedulerStatus(void *qhandle, void *tmrId);
void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *label, SSchedQueue *pSched) {
bool schedMalloced = false;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
bool schedMalloced = false;
if (NULL == pSched) {
pSched = (SSchedQueue *)taosMemoryCalloc(sizeof(SSchedQueue), 1);
if (pSched == NULL) {
@ -95,23 +97,32 @@ void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *lab
atomic_store_8(&pSched->stop, 0);
for (int32_t i = 0; i < numOfThreads; ++i) {
TdThreadAttr attr;
taosThreadAttrInit(&attr);
taosThreadAttrSetDetachState(&attr, PTHREAD_CREATE_JOINABLE);
int32_t code = taosThreadCreate(pSched->qthread + i, &attr, taosProcessSchedQueue, (void *)pSched);
taosThreadAttrDestroy(&attr);
if (code != 0) {
uError("%s: failed to create rpc thread(%s)", label, strerror(errno));
taosCleanUpScheduler(pSched);
if (schedMalloced) {
taosMemoryFree(pSched);
}
return NULL;
}
code = taosThreadAttrInit(&attr);
QUERY_CHECK_CODE(code, lino, _end);
code = taosThreadAttrSetDetachState(&attr, PTHREAD_CREATE_JOINABLE);
QUERY_CHECK_CODE(code, lino, _end);
code = taosThreadCreate(pSched->qthread + i, &attr, taosProcessSchedQueue, (void *)pSched);
QUERY_CHECK_CODE(code, lino, _end);
(void)taosThreadAttrDestroy(&attr);
++pSched->numOfThreads;
}
uDebug("%s scheduler is initialized, numOfThreads:%d", label, pSched->numOfThreads);
_end:
if (code != TSDB_CODE_SUCCESS) {
taosCleanUpScheduler(pSched);
if (schedMalloced) {
taosMemoryFree(pSched);
}
terrno = code;
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL;
}
return (void *)pSched;
}
@ -220,22 +231,22 @@ void taosCleanUpScheduler(void *param) {
for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
if (taosCheckPthreadValid(pSched->qthread[i])) {
tsem_post(&pSched->fullSem);
(void)tsem_post(&pSched->fullSem);
}
}
for (int32_t i = 0; i < pSched->numOfThreads; ++i) {
if (taosCheckPthreadValid(pSched->qthread[i])) {
taosThreadJoin(pSched->qthread[i], NULL);
(void)taosThreadJoin(pSched->qthread[i], NULL);
taosThreadClear(&pSched->qthread[i]);
}
}
tsem_destroy(&pSched->emptySem);
tsem_destroy(&pSched->fullSem);
taosThreadMutexDestroy(&pSched->queueMutex);
(void)tsem_destroy(&pSched->emptySem);
(void)tsem_destroy(&pSched->fullSem);
(void)taosThreadMutexDestroy(&pSched->queueMutex);
if (pSched->pTimer) {
taosTmrStop(pSched->pTimer);
(void)taosTmrStop(pSched->pTimer);
pSched->pTimer = NULL;
}

12
source/util/src/ttimer.c
View File

@ -516,7 +516,7 @@ static int32_t taosTmrModuleInit(void) {
tmrCtrls = taosMemoryMalloc(sizeof(tmr_ctrl_t) * tsMaxTmrCtrl);
if (tmrCtrls == NULL) {
tmrError("failed to allocate memory for timer controllers.");
return -1;
return terrno;
}
memset(&timerMap, 0, sizeof(timerMap));
@ -535,14 +535,14 @@ static int32_t taosTmrModuleInit(void) {
time_wheel_t* wheel = wheels + i;
if (taosThreadMutexInit(&wheel->mutex, NULL) != 0) {
tmrError("failed to create the mutex for wheel, reason:%s", strerror(errno));
return -1;
return terrno;
}
wheel->nextScanAt = now + wheel->resolution;
wheel->index = 0;
wheel->slots = (tmr_obj_t**)taosMemoryCalloc(wheel->size, sizeof(tmr_obj_t*));
if (wheel->slots == NULL) {
tmrError("failed to allocate wheel slots");
return -1;
return terrno;
}
timerMap.size += wheel->size;
}
@ -551,7 +551,7 @@ static int32_t taosTmrModuleInit(void) {
timerMap.slots = (timer_list_t*)taosMemoryCalloc(timerMap.size, sizeof(timer_list_t));
if (timerMap.slots == NULL) {
tmrError("failed to allocate hash map");
return -1;
return terrno;
}
tmrQhandle = taosInitScheduler(10000, taosTmrThreads, "tmr", NULL);
@ -570,7 +570,7 @@ static int32_t taosTmrInitModule(void) {
if (atomic_load_32(&tmrModuleInit) < 0) {
return -1;
}
while (true) {
if (0 == atomic_val_compare_exchange_32(&tmrModuleInit, 0, 1)) {
atomic_store_32(&tmrModuleInit, taosTmrModuleInit());
@ -609,7 +609,7 @@ void* taosTmrInit(int32_t maxNumOfTmrs, int32_t resolution, int32_t longest, con
}
tstrncpy(ctrl->label, label, sizeof(ctrl->label));
tmrDebug("%s timer controller is initialized, number of timer controllers: %d.", label, numOfTmrCtrl);
return ctrl;
}

57
source/util/src/tutil.c
View File

@ -107,6 +107,9 @@ char **strsplit(char *z, const char *delim, int32_t *num) {
int32_t size = 4;
char **split = taosMemoryMalloc(POINTER_BYTES * size);
if (split == NULL) {
return NULL;
}
for (char *p = strsep(&z, delim); p != NULL; p = strsep(&z, delim)) {
size_t len = strlen(p);
@ -118,7 +121,10 @@ char **strsplit(char *z, const char *delim, int32_t *num) {
if ((*num) >= size) {
size = (size << 1);
split = taosMemoryRealloc(split, POINTER_BYTES * size);
ASSERTS(NULL != split, "realloc memory failed. size=%d", (int32_t) POINTER_BYTES * size);
if (split == NULL) {
return NULL;
}
ASSERTS(NULL != split, "realloc memory failed. size=%d", (int32_t)POINTER_BYTES * size);
}
}
@ -145,10 +151,10 @@ char *strnchr(const char *haystack, char needle, int32_t len, bool skipquote) {
return NULL;
}
TdUcs4* wcsnchr(const TdUcs4* haystack, TdUcs4 needle, size_t len) {
for(int32_t i = 0; i < len; ++i) {
TdUcs4 *wcsnchr(const TdUcs4 *haystack, TdUcs4 needle, size_t len) {
for (int32_t i = 0; i < len; ++i) {
if (haystack[i] == needle) {
return (TdUcs4*) &haystack[i];
return (TdUcs4 *)&haystack[i];
}
}
@ -314,6 +320,9 @@ char *strbetween(char *string, char *begin, char *end) {
int32_t size = (int32_t)(_end - _begin);
if (_end != NULL && size > 0) {
result = (char *)taosMemoryCalloc(1, size);
if (result) {
return NULL;
}
memcpy(result, _begin + strlen(begin), size - +strlen(begin));
}
}
@ -324,13 +333,13 @@ int32_t tintToHex(uint64_t val, char hex[]) {
const char hexstr[16] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};
int32_t j = 0, k = 0;
if (val == 0) {
if (val == 0) {
hex[j++] = hexstr[0];
return j;
}
// ignore the initial 0
while((val & (((uint64_t)0xfL) << ((15 - k) * 4))) == 0) {
while ((val & (((uint64_t)0xfL) << ((15 - k) * 4))) == 0) {
k += 1;
}
@ -346,10 +355,10 @@ int32_t titoa(uint64_t val, size_t radix, char str[]) {
return 0;
}
const char* s = "0123456789abcdef";
char buf[65] = {0};
const char *s = "0123456789abcdef";
char buf[65] = {0};
int32_t i = 0;
int32_t i = 0;
uint64_t v = val;
do {
buf[i++] = s[v % radix];
@ -357,7 +366,7 @@ int32_t titoa(uint64_t val, size_t radix, char str[]) {
} while (v > 0);
// reverse order
for(int32_t j = 0; j < i; ++j) {
for (int32_t j = 0; j < i; ++j) {
str[j] = buf[i - j - 1];
}
@ -429,8 +438,8 @@ void taosIpPort2String(uint32_t ip, uint16_t port, char *str) {
size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize) {
if (rsize == 0 || rsize == 1) {
char* p = strnchr(str, reject[0], size, false);
return (p == NULL)? size:(p-str);
char *p = strnchr(str, reject[0], size, false);
return (p == NULL) ? size : (p - str);
}
/* Use multiple small memsets to enable inlining on most targets. */
@ -441,15 +450,15 @@ size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize)
memset(p + 192, 0, 64);
unsigned char *s = (unsigned char *)reject;
int32_t index = 0;
int32_t index = 0;
do {
p[s[index++]] = 1;
} while (index < rsize);
s = (unsigned char*) str;
s = (unsigned char *)str;
int32_t times = size >> 2;
if (times == 0) {
for(int32_t i = 0; i < size; ++i) {
for (int32_t i = 0; i < size; ++i) {
if (p[s[i]]) {
return i;
}
@ -460,7 +469,7 @@ size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize)
index = 0;
uint32_t c0, c1, c2, c3;
for(int32_t i = 0; i < times; ++i, index += 4) {
for (int32_t i = 0; i < times; ++i, index += 4) {
int32_t j = index;
c0 = p[s[j]];
c1 = p[s[j + 1]];
@ -474,7 +483,7 @@ size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize)
}
int32_t offset = times * 4;
for(int32_t i = offset; i < size; ++i) {
for (int32_t i = offset; i < size; ++i) {
if (p[s[i]]) {
return i;
}
@ -485,8 +494,8 @@ size_t tstrncspn(const char *str, size_t size, const char *reject, size_t rsize)
size_t twcsncspn(const TdUcs4 *wcs, size_t size, const TdUcs4 *reject, size_t rsize) {
if (rsize == 0 || rsize == 1) {
TdUcs4* p = wcsnchr(wcs, reject[0], size);
return (p == NULL)? size:(p-wcs);
TdUcs4 *p = wcsnchr(wcs, reject[0], size);
return (p == NULL) ? size : (p - wcs);
}
size_t index = 0;
@ -497,19 +506,17 @@ size_t twcsncspn(const TdUcs4 *wcs, size_t size, const TdUcs4 *reject, size_t rs
return index;
}
int32_t parseCfgReal(const char* str, double* out) {
int32_t parseCfgReal(const char *str, double *out) {
double val;
char *endPtr;
errno = 0;
val = taosStr2Double(str, &endPtr);
if (str == endPtr || errno == ERANGE || isnan(val)) {
terrno = TSDB_CODE_INVALID_CFG_VALUE;
return -1;
return terrno = TSDB_CODE_INVALID_CFG_VALUE;
}
while(isspace((unsigned char)*endPtr)) endPtr++;
while (isspace((unsigned char)*endPtr)) endPtr++;
if (*endPtr != '\0') {
terrno = TSDB_CODE_INVALID_CFG_VALUE;
return -1;
return terrno = TSDB_CODE_INVALID_CFG_VALUE;
}
*out = val;
return TSDB_CODE_SUCCESS;