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

Merge pull request #24515 from taosdata/fix/fix/TD-28334-3.0 

clone task id
This commit is contained in:
dapan1121 2024-01-18 14:42:57 +08:00 committed by GitHub
parent 567aab933d 7986951f44
commit d44d945b09
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/executor/inc/executorInt.h
View File

@ -200,6 +200,7 @@ typedef struct SExchangeInfo {
uint64_t self; uint64_t self;
SLimitInfo limitInfo; SLimitInfo limitInfo;
int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo
char* pTaskId;
} SExchangeInfo; } SExchangeInfo;
typedef struct SScanInfo { typedef struct SScanInfo {

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

@ -260,14 +260,17 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t len = strlen(id) + 1;
pInfo->pTaskId = taosMemoryCalloc(1, len);
strncpy(pInfo->pTaskId, id, len);
for (int32_t i = 0; i < numOfSources; ++i) { for (int32_t i = 0; i < numOfSources; ++i) {
SSourceDataInfo dataInfo = {0}; SSourceDataInfo dataInfo = {0};
dataInfo.status = EX_SOURCE_DATA_NOT_READY; dataInfo.status = EX_SOURCE_DATA_NOT_READY;
dataInfo.taskId = id; dataInfo.taskId = pInfo->pTaskId;
dataInfo.index = i; dataInfo.index = i;
SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo);
if (pDs == NULL) { if (pDs == NULL) {
taosArrayDestroy(pInfo->pSourceDataInfo); taosArrayDestroyEx(pInfo->pSourceDataInfo, freeSourceDataInfo);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
} }
@ -383,6 +386,8 @@ void doDestroyExchangeOperatorInfo(void* param) {
tSimpleHashCleanup(pExInfo->pHashSources); tSimpleHashCleanup(pExInfo->pHashSources);
tsem_destroy(&pExInfo->ready); tsem_destroy(&pExInfo->ready);
taosMemoryFreeClear(pExInfo->pTaskId);
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
@ -782,7 +787,7 @@ int32_t addSingleExchangeSource(SOperatorInfo* pOperator, SExchangeOperatorBasic
if (pIdx->inUseIdx < 0) { if (pIdx->inUseIdx < 0) {
SSourceDataInfo dataInfo = {0}; SSourceDataInfo dataInfo = {0};
dataInfo.status = EX_SOURCE_DATA_NOT_READY; dataInfo.status = EX_SOURCE_DATA_NOT_READY;
dataInfo.taskId = GET_TASKID(pOperator->pTaskInfo); dataInfo.taskId = pExchangeInfo->pTaskId;
dataInfo.index = pIdx->srcIdx; dataInfo.index = pIdx->srcIdx;
dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL); dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL);
dataInfo.srcOpType = pBasicParam->srcOpType; dataInfo.srcOpType = pBasicParam->srcOpType;