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

use taskid of operator

This commit is contained in:
54liuyao 2024-01-18 09:22:39 +08:00
parent 1309e56a95
commit 7986951f44
2 changed files with 8 additions and 8 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 {

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

@ -37,7 +37,7 @@ typedef struct SSourceDataInfo {
int64_t startTime; int64_t startTime;
int32_t code; int32_t code;
EX_SOURCE_STATUS status; EX_SOURCE_STATUS status;
char* taskId; const char* taskId;
SArray* pSrcUidList; SArray* pSrcUidList;
int32_t srcOpType; int32_t srcOpType;
bool tableSeq; bool tableSeq;
@ -261,11 +261,12 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const
} }
int32_t len = strlen(id) + 1; 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 = taosMemoryCalloc(1, len); dataInfo.taskId = pInfo->pTaskId;
strncpy(dataInfo.taskId, id, len);
dataInfo.index = i; dataInfo.index = i;
SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo);
if (pDs == NULL) { if (pDs == NULL) {
@ -370,7 +371,6 @@ void freeBlock(void* pParam) {
void freeSourceDataInfo(void* p) { void freeSourceDataInfo(void* p) {
SSourceDataInfo* pInfo = (SSourceDataInfo*)p; SSourceDataInfo* pInfo = (SSourceDataInfo*)p;
taosMemoryFreeClear(pInfo->pRsp); taosMemoryFreeClear(pInfo->pRsp);
taosMemoryFreeClear(pInfo->taskId);
} }
void doDestroyExchangeOperatorInfo(void* param) { void doDestroyExchangeOperatorInfo(void* param) {
@ -386,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);
} }
@ -785,10 +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;
char* pTaskId = GET_TASKID(pOperator->pTaskInfo); dataInfo.taskId = pExchangeInfo->pTaskId;
int32_t len = strlen(pTaskId) + 1;
dataInfo.taskId = taosMemoryCalloc(1, len);
strncpy(dataInfo.taskId, pTaskId, len);
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;