From b48752686e489a86d2597b91e4d8dcde16616efb Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Wed, 9 Nov 2022 13:45:46 +0800 Subject: [PATCH] refactor: do some internal refactor. --- include/common/tdatablock.h | 2 +- source/client/src/clientMsgHandler.c | 3 +- source/common/src/tdatablock.c | 31 +- source/dnode/mgmt/mgmt_dnode/src/dmHandle.c | 3 +- source/dnode/mnode/impl/src/mndShow.c | 3 +- source/dnode/vnode/src/tq/tqExec.c | 4 +- source/libs/command/src/command.c | 3 +- source/libs/command/src/explain.c | 3 +- source/libs/executor/inc/executorimpl.h | 8 +- source/libs/executor/src/cachescanoperator.c | 2 +- source/libs/executor/src/dataDispatcher.c | 2 +- source/libs/executor/src/executor.c | 21 + source/libs/executor/src/executorimpl.c | 628 +----------------- source/libs/executor/src/groupoperator.c | 6 +- source/libs/executor/src/joinoperator.c | 7 +- source/libs/executor/src/projectoperator.c | 5 +- source/libs/executor/src/scanoperator.c | 17 +- source/libs/executor/src/sortoperator.c | 6 +- source/libs/executor/src/tfill.c | 2 +- source/libs/executor/src/timewindowoperator.c | 22 +- source/libs/stream/src/streamDispatch.c | 6 +- 21 files changed, 87 insertions(+), 697 deletions(-) diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h index fdbb2c5123..502ba10d33 100644 --- a/include/common/tdatablock.h +++ b/include/common/tdatablock.h @@ -244,7 +244,7 @@ int32_t blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColIn SColumnInfoData createColumnInfoData(int16_t type, int32_t bytes, int16_t colId); SColumnInfoData* bdGetColumnInfoData(const SSDataBlock* pBlock, int32_t index); -void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols, int8_t needCompress); +int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols); const char* blockDecode(SSDataBlock* pBlock, const char* pData); void blockDebugShowDataBlock(SSDataBlock* pBlock, const char* flag); diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index e17e7e79c6..591d469d7c 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -442,8 +442,7 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) { (*pRsp)->numOfRows = htonl(pBlock->info.rows); (*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS); - int32_t len = 0; - blockEncode(pBlock, (*pRsp)->data, &len, SHOW_VARIABLES_RESULT_COLS, false); + int32_t len = blockEncode(pBlock, (*pRsp)->data, SHOW_VARIABLES_RESULT_COLS); ASSERT(len == rspSize - sizeof(SRetrieveTableRsp)); blockDataDestroy(pBlock); diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 5d0121aa15..536cbed33e 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -2197,7 +2197,9 @@ char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) { return rname.ctbShortName; } -void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols, int8_t needCompress) { +int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) { + int32_t dataLen = 0; + // todo extract method int32_t* version = (int32_t*)data; *version = 1; @@ -2238,7 +2240,7 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_ int32_t* colSizes = (int32_t*)data; data += numOfCols * sizeof(int32_t); - *dataLen = blockDataGetSerialMetaSize(numOfCols); + dataLen = blockDataGetSerialMetaSize(numOfCols); int32_t numOfRows = pBlock->info.rows; for (int32_t col = 0; col < numOfCols; ++col) { @@ -2255,26 +2257,23 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_ } data += metaSize; - (*dataLen) += metaSize; + dataLen += metaSize; - if (needCompress) { - colSizes[col] = blockCompressColData(pColRes, numOfRows, data, needCompress); - data += colSizes[col]; - (*dataLen) += colSizes[col]; - } else { - colSizes[col] = colDataGetLength(pColRes, numOfRows); - (*dataLen) += colSizes[col]; - memmove(data, pColRes->pData, colSizes[col]); - data += colSizes[col]; - } + colSizes[col] = colDataGetLength(pColRes, numOfRows); + dataLen += colSizes[col]; + memmove(data, pColRes->pData, colSizes[col]); + data += colSizes[col]; colSizes[col] = htonl(colSizes[col]); } - *actualLen = *dataLen; + *actualLen = dataLen; *groupId = pBlock->info.groupId; - ASSERT(*dataLen > 0); - uDebug("build data block, actualLen:%d, rows:%d, cols:%d", *dataLen, *rows, *cols); + ASSERT(dataLen > 0); + + uDebug("build data block, actualLen:%d, rows:%d, cols:%d", dataLen, *rows, *cols); + + return dataLen; } const char* blockDecode(SSDataBlock* pBlock, const char* pData) { diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c index 85a09b79fd..8a8561161b 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmHandle.c @@ -307,8 +307,7 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) { pStart += sizeof(SSysTableSchema); } - int32_t len = 0; - blockEncode(pBlock, pStart, &len, numOfCols, false); + int32_t len = blockEncode(pBlock, pStart, numOfCols); pRsp->numOfRows = htonl(pBlock->info.rows); pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision diff --git a/source/dnode/mnode/impl/src/mndShow.c b/source/dnode/mnode/impl/src/mndShow.c index 637a3818e6..b0af98b933 100644 --- a/source/dnode/mnode/impl/src/mndShow.c +++ b/source/dnode/mnode/impl/src/mndShow.c @@ -303,8 +303,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) { pStart += sizeof(SSysTableSchema); } - int32_t len = 0; - blockEncode(pBlock, pStart, &len, pShow->pMeta->numOfColumns, false); + int32_t len = blockEncode(pBlock, pStart, pShow->pMeta->numOfColumns); } pRsp->numOfRows = htonl(rowsRead); diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index a6e8767a4d..48c14bc758 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -27,9 +27,7 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t pRetrieve->completed = 1; pRetrieve->numOfRows = htonl(pBlock->info.rows); - // TODO enable compress - int32_t actualLen = 0; - blockEncode(pBlock, pRetrieve->data, &actualLen, numOfCols, false); + int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols); actualLen += sizeof(SRetrieveTableRsp); ASSERT(actualLen <= dataStrLen); taosArrayPush(pRsp->blockDataLen, &actualLen); diff --git a/source/libs/command/src/command.c b/source/libs/command/src/command.c index 76c84b3be9..1c2d7e1f66 100644 --- a/source/libs/command/src/command.c +++ b/source/libs/command/src/command.c @@ -39,8 +39,7 @@ static int32_t buildRetrieveTableRsp(SSDataBlock* pBlock, int32_t numOfCols, SRe (*pRsp)->numOfRows = htonl(pBlock->info.rows); (*pRsp)->numOfCols = htonl(numOfCols); - int32_t len = 0; - blockEncode(pBlock, (*pRsp)->data, &len, numOfCols, false); + int32_t len = blockEncode(pBlock, (*pRsp)->data, numOfCols); ASSERT(len == rspSize - sizeof(SRetrieveTableRsp)); return TSDB_CODE_SUCCESS; diff --git a/source/libs/command/src/explain.c b/source/libs/command/src/explain.c index 80a524496c..915dc08c14 100644 --- a/source/libs/command/src/explain.c +++ b/source/libs/command/src/explain.c @@ -1610,8 +1610,7 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) { rsp->completed = 1; rsp->numOfRows = htonl(rowNum); - int32_t len = 0; - blockEncode(pBlock, rsp->data, &len, taosArrayGetSize(pBlock->pDataBlock), 0); + int32_t len = blockEncode(pBlock, rsp->data, taosArrayGetSize(pBlock->pDataBlock)); ASSERT(len == rspSize - sizeof(SRetrieveTableRsp)); rsp->compLen = htonl(len); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index ced668cf37..77915624dd 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -842,10 +842,8 @@ typedef struct SJoinOperatorInfo { #define OPTR_IS_OPENED(_optr) (((_optr)->status & OP_OPENED) == OP_OPENED) #define OPTR_SET_OPENED(_optr) ((_optr)->status |= OP_OPENED) -void doDestroyExchangeOperatorInfo(void* param); - -SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t streamFn, - __optr_fn_t cleanup, __optr_close_fn_t closeFn, __optr_explain_fn_t explain); +SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup, + __optr_close_fn_t closeFn, __optr_explain_fn_t explain); int32_t operatorDummyOpenFn(SOperatorInfo* pOperator); int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num); @@ -881,6 +879,8 @@ STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInter int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag); int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz); +void doDestroyExchangeOperatorInfo(void* param); + void doSetOperatorCompleted(SOperatorInfo* pOperator); void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo); int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, diff --git a/source/libs/executor/src/cachescanoperator.c b/source/libs/executor/src/cachescanoperator.c index 76866eedb7..41fcc44f97 100644 --- a/source/libs/executor/src/cachescanoperator.c +++ b/source/libs/executor/src/cachescanoperator.c @@ -102,7 +102,7 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock); pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doScanCache, NULL, NULL, destroyLastrowScanOperator, NULL); + createOperatorFpSet(operatorDummyOpenFn, doScanCache, NULL, destroyLastrowScanOperator, NULL); pOperator->cost.openCost = 0; return pOperator; diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index bc4ab9c468..3d9757c96f 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -76,7 +76,7 @@ static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pIn pEntry->dataLen = 0; pBuf->useSize = sizeof(SDataCacheEntry); - blockEncode(pInput->pData, pEntry->data, &pEntry->dataLen, numOfCols, pEntry->compressed); + pEntry->dataLen = blockEncode(pInput->pData, pEntry->data, numOfCols); ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 1cf01c8661..1aa9a3c613 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -1106,3 +1106,24 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT } return 0; } + +void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { + SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle; + assert(pMsg->info.ahandle != NULL); + + SDataBuf buf = {.len = pMsg->contLen, .pData = NULL}; + + if (pMsg->contLen > 0) { + buf.pData = taosMemoryCalloc(1, pMsg->contLen); + if (buf.pData == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + pMsg->code = TSDB_CODE_OUT_OF_MEMORY; + } else { + memcpy(buf.pData, pMsg->pCont, pMsg->contLen); + } + } + + pSendInfo->fp(pSendInfo->param, &buf, pMsg->code); + rpcFreeCont(pMsg->pCont); + destroySendMsgInfo(pSendInfo); +} \ No newline at end of file diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 7d662f8784..ca1d4cccab 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -32,7 +32,6 @@ #include "index.h" #include "query.h" #include "tcompare.h" -#include "tcompression.h" #include "thash.h" #include "ttypes.h" #include "vnode.h" @@ -91,7 +90,7 @@ static void destroySortOperatorInfo(void* param); static void destroyAggOperatorInfo(void* param); static void destroyIntervalOperatorInfo(void* param); -static void destroyExchangeOperatorInfo(void* param); + static void destroyOperatorInfo(SOperatorInfo* pOperator); @@ -109,8 +108,8 @@ int32_t operatorDummyOpenFn(SOperatorInfo* pOperator) { return TSDB_CODE_SUCCESS; } -SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t streamFn, - __optr_fn_t cleanup, __optr_close_fn_t closeFn, __optr_explain_fn_t explain) { +SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup, + __optr_close_fn_t closeFn, __optr_explain_fn_t explain) { SOperatorFpSet fpSet = { ._openFn = openFn, .getNextFn = nextFn, @@ -1652,596 +1651,6 @@ int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t return TSDB_CODE_SUCCESS; } -typedef struct SFetchRspHandleWrapper { - uint32_t exchangeId; - int32_t sourceIndex; -} SFetchRspHandleWrapper; - -int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) { - SFetchRspHandleWrapper* pWrapper = (SFetchRspHandleWrapper*)param; - - SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId); - if (pExchangeInfo == NULL) { - qWarn("failed to acquire exchange operator, since it may have been released"); - taosMemoryFree(pMsg->pData); - return TSDB_CODE_SUCCESS; - } - - int32_t index = pWrapper->sourceIndex; - SSourceDataInfo* pSourceDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, index); - - if (code == TSDB_CODE_SUCCESS) { - pSourceDataInfo->pRsp = pMsg->pData; - - SRetrieveTableRsp* pRsp = pSourceDataInfo->pRsp; - pRsp->numOfRows = htonl(pRsp->numOfRows); - pRsp->compLen = htonl(pRsp->compLen); - pRsp->numOfCols = htonl(pRsp->numOfCols); - pRsp->useconds = htobe64(pRsp->useconds); - pRsp->numOfBlocks = htonl(pRsp->numOfBlocks); - - ASSERT(pRsp != NULL); - qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d", pSourceDataInfo->taskId, index, pRsp->numOfBlocks, - pRsp->numOfRows); - } else { - taosMemoryFree(pMsg->pData); - pSourceDataInfo->code = code; - qDebug("%s fetch rsp received, index:%d, error:%s", pSourceDataInfo->taskId, index, tstrerror(code)); - } - - pSourceDataInfo->status = EX_SOURCE_DATA_READY; - - tsem_post(&pExchangeInfo->ready); - taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId); - - return TSDB_CODE_SUCCESS; -} - -void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { - SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle; - assert(pMsg->info.ahandle != NULL); - - SDataBuf buf = {.len = pMsg->contLen, .pData = NULL}; - - if (pMsg->contLen > 0) { - buf.pData = taosMemoryCalloc(1, pMsg->contLen); - if (buf.pData == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - pMsg->code = TSDB_CODE_OUT_OF_MEMORY; - } else { - memcpy(buf.pData, pMsg->pCont, pMsg->contLen); - } - } - - pSendInfo->fp(pSendInfo->param, &buf, pMsg->code); - rpcFreeCont(pMsg->pCont); - destroySendMsgInfo(pSendInfo); -} - -static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) { - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); - - SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, sourceIndex); - SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, sourceIndex); - - ASSERT(pDataInfo->status == EX_SOURCE_DATA_NOT_READY); - - SFetchRspHandleWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SFetchRspHandleWrapper)); - pWrapper->exchangeId = pExchangeInfo->self; - pWrapper->sourceIndex = sourceIndex; - - if (pSource->localExec) { - SDataBuf pBuf = {0}; - int32_t code = - (*pTaskInfo->localFetch.fp)(pTaskInfo->localFetch.handle, pSource->schedId, pTaskInfo->id.queryId, - pSource->taskId, 0, pSource->execId, &pBuf.pData, pTaskInfo->localFetch.explainRes); - loadRemoteDataCallback(pWrapper, &pBuf, code); - taosMemoryFree(pWrapper); - } else { - SResFetchReq* pMsg = taosMemoryCalloc(1, sizeof(SResFetchReq)); - if (NULL == pMsg) { - pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY; - taosMemoryFree(pWrapper); - return pTaskInfo->code; - } - - qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu, - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId, - pSource->execId, sourceIndex, totalSources); - - pMsg->header.vgId = htonl(pSource->addr.nodeId); - pMsg->sId = htobe64(pSource->schedId); - pMsg->taskId = htobe64(pSource->taskId); - pMsg->queryId = htobe64(pTaskInfo->id.queryId); - pMsg->execId = htonl(pSource->execId); - - // send the fetch remote task result reques - SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); - if (NULL == pMsgSendInfo) { - taosMemoryFreeClear(pMsg); - taosMemoryFree(pWrapper); - qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo)); - pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY; - return pTaskInfo->code; - } - - pMsgSendInfo->param = pWrapper; - pMsgSendInfo->paramFreeFp = taosMemoryFree; - pMsgSendInfo->msgInfo.pData = pMsg; - pMsgSendInfo->msgInfo.len = sizeof(SResFetchReq); - pMsgSendInfo->msgType = pSource->fetchMsgType; - pMsgSendInfo->fp = loadRemoteDataCallback; - - int64_t transporterId = 0; - int32_t code = - asyncSendMsgToServer(pExchangeInfo->pTransporter, &pSource->addr.epSet, &transporterId, pMsgSendInfo); - } - - return TSDB_CODE_SUCCESS; -} - -void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs, - SOperatorInfo* pOperator) { - pInfo->totalRows += numOfRows; - pInfo->totalSize += dataLen; - pInfo->totalElapsed += (taosGetTimestampUs() - startTs); - pOperator->resultInfo.totalRows += numOfRows; -} - -int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) { - if (pColList == NULL) { // data from other sources - blockDataCleanup(pRes); - *pNextStart = (char*)blockDecode(pRes, pData); - } else { // extract data according to pColList - char* pStart = pData; - - int32_t numOfCols = htonl(*(int32_t*)pStart); - pStart += sizeof(int32_t); - - // todo refactor:extract method - SSysTableSchema* pSchema = (SSysTableSchema*)pStart; - for (int32_t i = 0; i < numOfCols; ++i) { - SSysTableSchema* p = (SSysTableSchema*)pStart; - - p->colId = htons(p->colId); - p->bytes = htonl(p->bytes); - pStart += sizeof(SSysTableSchema); - } - - SSDataBlock* pBlock = createDataBlock(); - for (int32_t i = 0; i < numOfCols; ++i) { - SColumnInfoData idata = createColumnInfoData(pSchema[i].type, pSchema[i].bytes, pSchema[i].colId); - blockDataAppendColInfo(pBlock, &idata); - } - - blockDecode(pBlock, pStart); - blockDataEnsureCapacity(pRes, pBlock->info.rows); - - // data from mnode - pRes->info.rows = pBlock->info.rows; - relocateColumnData(pRes, pColList, pBlock->pDataBlock, false); - blockDataDestroy(pBlock); - } - - // todo move this to time window aggregator, since the primary timestamp may not be known by exchange operator. - blockDataUpdateTsWindow(pRes, 0); - return TSDB_CODE_SUCCESS; -} - -static void* setAllSourcesCompleted(SOperatorInfo* pOperator, int64_t startTs) { - SExchangeInfo* pExchangeInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - int64_t el = taosGetTimestampUs() - startTs; - SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; - - pLoadInfo->totalElapsed += el; - - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); - qDebug("%s all %" PRIzu " sources are exhausted, total rows: %" PRIu64 " bytes:%" PRIu64 ", elapsed:%.2f ms", - GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize, - pLoadInfo->totalElapsed / 1000.0); - - doSetOperatorCompleted(pOperator); - return NULL; -} - - -static int32_t getCompletedSources(const SArray* pArray) { - size_t total = taosArrayGetSize(pArray); - - int32_t completed = 0; - for (int32_t k = 0; k < total; ++k) { - SSourceDataInfo* p = taosArrayGet(pArray, k); - if (p->status == EX_SOURCE_DATA_EXHAUSTED) { - completed += 1; - } - } - - return completed; -} - -static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo, - SExecTaskInfo* pTaskInfo) { - int32_t code = 0; - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSourceDataInfo); - int32_t completed = getCompletedSources(pExchangeInfo->pSourceDataInfo); - if (completed == totalSources) { - setAllSourcesCompleted(pOperator, pExchangeInfo->openedTs); - return; - } - - while (1) { - tsem_wait(&pExchangeInfo->ready); - - for (int32_t i = 0; i < totalSources; ++i) { - SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i); - if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) { - continue; - } - - if (pDataInfo->status != EX_SOURCE_DATA_READY) { - continue; - } - - if (pDataInfo->code != TSDB_CODE_SUCCESS) { - code = pDataInfo->code; - goto _error; - } - - SRetrieveTableRsp* pRsp = pDataInfo->pRsp; - SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, i); - - // todo - SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; - if (pRsp->numOfRows == 0) { - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; - qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64 - ", totalRows:%" PRIu64 ", try next %d/%" PRIzu, - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows, - pExchangeInfo->loadInfo.totalRows, i + 1, totalSources); - taosMemoryFreeClear(pDataInfo->pRsp); - break; - } - - SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp; - int32_t index = 0; - char* pStart = pRetrieveRsp->data; - while (index++ < pRetrieveRsp->numOfBlocks) { - SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false); - code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart); - if (code != 0) { - taosMemoryFreeClear(pDataInfo->pRsp); - goto _error; - } - - taosArrayPush(pExchangeInfo->pResultBlockList, &pb); - } - - updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, pExchangeInfo->openedTs, pOperator); - - if (pRsp->completed == 1) { - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; - qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 - " execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 - ", total:%.2f Kb, try next %d/%" PRIzu, - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks, - pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0, - i + 1, totalSources); - } else { - qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 - " execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb", - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRsp->numOfBlocks, - pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0); - } - - taosMemoryFreeClear(pDataInfo->pRsp); - - if (pDataInfo->status != EX_SOURCE_DATA_EXHAUSTED) { - pDataInfo->status = EX_SOURCE_DATA_NOT_READY; - code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i); - if (code != TSDB_CODE_SUCCESS) { - taosMemoryFreeClear(pDataInfo->pRsp); - goto _error; - } - } - return; - } // end loop - - int32_t complete1 = getCompletedSources(pExchangeInfo->pSourceDataInfo); - if (complete1 == totalSources) { - qDebug("all sources are completed, %s", GET_TASKID(pTaskInfo)); - return; - } - } - -_error: - pTaskInfo->code = code; -} - -static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) { - SExchangeInfo* pExchangeInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); - int64_t startTs = taosGetTimestampUs(); - - // Asynchronously send all fetch requests to all sources. - for (int32_t i = 0; i < totalSources; ++i) { - int32_t code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i); - if (code != TSDB_CODE_SUCCESS) { - pTaskInfo->code = code; - return code; - } - } - - int64_t endTs = taosGetTimestampUs(); - qDebug("%s send all fetch requests to %" PRIzu " sources completed, elapsed:%.2fms", GET_TASKID(pTaskInfo), - totalSources, (endTs - startTs) / 1000.0); - - pOperator->status = OP_RES_TO_RETURN; - pOperator->cost.openCost = taosGetTimestampUs() - startTs; - - tsem_wait(&pExchangeInfo->ready); - tsem_post(&pExchangeInfo->ready); - return TSDB_CODE_SUCCESS; -} - -static int32_t seqLoadRemoteData(SOperatorInfo* pOperator) { - SExchangeInfo* pExchangeInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); - int64_t startTs = taosGetTimestampUs(); - - while (1) { - if (pExchangeInfo->current >= totalSources) { - setAllSourcesCompleted(pOperator, startTs); - return TSDB_CODE_SUCCESS; - } - - doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current); - tsem_wait(&pExchangeInfo->ready); - - SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current); - SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current); - - if (pDataInfo->code != TSDB_CODE_SUCCESS) { - qError("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d error happens, code:%s", GET_TASKID(pTaskInfo), - pSource->addr.nodeId, pSource->taskId, pSource->execId, tstrerror(pDataInfo->code)); - pOperator->pTaskInfo->code = pDataInfo->code; - return pOperator->pTaskInfo->code; - } - - SRetrieveTableRsp* pRsp = pDataInfo->pRsp; - SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; - if (pRsp->numOfRows == 0) { - qDebug("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d %d of total completed, rowsOfSource:%" PRIu64 - ", totalRows:%" PRIu64 " try next", - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pExchangeInfo->current + 1, - pDataInfo->totalRows, pLoadInfo->totalRows); - - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; - pExchangeInfo->current += 1; - taosMemoryFreeClear(pDataInfo->pRsp); - continue; - } - - SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp; - - char* pStart = pRetrieveRsp->data; - int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, NULL, &pStart); - - if (pRsp->completed == 1) { - qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64 - ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu, - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows, - pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize, pExchangeInfo->current + 1, - totalSources); - - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; - pExchangeInfo->current += 1; - } else { - qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, totalRows:%" PRIu64 - ", totalBytes:%" PRIu64, - GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows, - pLoadInfo->totalRows, pLoadInfo->totalSize); - } - - updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator); - pDataInfo->totalRows += pRetrieveRsp->numOfRows; - - taosMemoryFreeClear(pDataInfo->pRsp); - return TSDB_CODE_SUCCESS; - } -} - -static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) { - if (OPTR_IS_OPENED(pOperator)) { - return TSDB_CODE_SUCCESS; - } - - int64_t st = taosGetTimestampUs(); - - SExchangeInfo* pExchangeInfo = pOperator->info; - if (!pExchangeInfo->seqLoadData) { - int32_t code = prepareConcurrentlyLoad(pOperator); - if (code != TSDB_CODE_SUCCESS) { - return code; - } - pExchangeInfo->openedTs = taosGetTimestampUs(); - } - - OPTR_SET_OPENED(pOperator); - pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0; - return TSDB_CODE_SUCCESS; -} - -static void freeBlock(void* pParam) { - SSDataBlock* pBlock = *(SSDataBlock**)pParam; - blockDataDestroy(pBlock); -} - -static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) { - SExchangeInfo* pExchangeInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - pTaskInfo->code = pOperator->fpSet._openFn(pOperator); - if (pTaskInfo->code != TSDB_CODE_SUCCESS) { - return NULL; - } - - size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); - - SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; - if (pOperator->status == OP_EXEC_DONE) { - qDebug("%s all %" PRIzu " source(s) are exhausted, total rows:%" PRIu64 " bytes:%" PRIu64 ", elapsed:%.2f ms", - GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize, - pLoadInfo->totalElapsed / 1000.0); - return NULL; - } - - size_t size = taosArrayGetSize(pExchangeInfo->pResultBlockList); - if (size == 0 || pExchangeInfo->rspBlockIndex >= size) { - pExchangeInfo->rspBlockIndex = 0; - taosArrayClearEx(pExchangeInfo->pResultBlockList, freeBlock); - if (pExchangeInfo->seqLoadData) { - seqLoadRemoteData(pOperator); - } else { - concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo); - } - - if (taosArrayGetSize(pExchangeInfo->pResultBlockList) == 0) { - return NULL; - } - } - - // we have buffered retrieved datablock, return it directly - return taosArrayGetP(pExchangeInfo->pResultBlockList, pExchangeInfo->rspBlockIndex++); -} - -static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) { - SExchangeInfo* pExchangeInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - if (pOperator->status == OP_EXEC_DONE) { - return NULL; - } - - while (1) { - SSDataBlock* pBlock = doLoadRemoteDataImpl(pOperator); - if (pBlock == NULL) { - return NULL; - } - - SLimitInfo* pLimitInfo = &pExchangeInfo->limitInfo; - if (hasLimitOffsetInfo(pLimitInfo)) { - int32_t status = handleLimitOffset(pOperator, pLimitInfo, pBlock, false); - if (status == PROJECT_RETRIEVE_CONTINUE) { - continue; - } else if (status == PROJECT_RETRIEVE_DONE) { - size_t rows = pBlock->info.rows; - pExchangeInfo->limitInfo.numOfOutputRows += rows; - - if (rows == 0) { - doSetOperatorCompleted(pOperator); - return NULL; - } else { - return pBlock; - } - } - } else { - return pBlock; - } - } -} - -static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const char* id) { - pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo)); - if (pInfo->pSourceDataInfo == NULL) { - return TSDB_CODE_OUT_OF_MEMORY; - } - - for (int32_t i = 0; i < numOfSources; ++i) { - SSourceDataInfo dataInfo = {0}; - dataInfo.status = EX_SOURCE_DATA_NOT_READY; - dataInfo.taskId = id; - dataInfo.index = i; - SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); - if (pDs == NULL) { - taosArrayDestroy(pInfo->pSourceDataInfo); - return TSDB_CODE_OUT_OF_MEMORY; - } - } - - return TSDB_CODE_SUCCESS; -} - -static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo, const char* id) { - size_t numOfSources = LIST_LENGTH(pExNode->pSrcEndPoints); - - if (numOfSources == 0) { - qError("%s invalid number: %d of sources in exchange operator", id, (int32_t)numOfSources); - return TSDB_CODE_INVALID_PARA; - } - - pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode)); - if (pInfo->pSources == NULL) { - return TSDB_CODE_OUT_OF_MEMORY; - } - - for (int32_t i = 0; i < numOfSources; ++i) { - SDownstreamSourceNode* pNode = (SDownstreamSourceNode*)nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i); - taosArrayPush(pInfo->pSources, pNode); - } - - initLimitInfo(pExNode->node.pLimit, pExNode->node.pSlimit, &pInfo->limitInfo); - pInfo->self = taosAddRef(exchangeObjRefPool, pInfo); - - return initDataSource(numOfSources, pInfo, id); -} - -SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo) { - SExchangeInfo* pInfo = taosMemoryCalloc(1, sizeof(SExchangeInfo)); - SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - if (pInfo == NULL || pOperator == NULL) { - goto _error; - } - - int32_t code = initExchangeOperator(pExNode, pInfo, GET_TASKID(pTaskInfo)); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } - - tsem_init(&pInfo->ready, 0, 0); - pInfo->pDummyBlock = createResDataBlock(pExNode->node.pOutputDataBlockDesc); - pInfo->pResultBlockList = taosArrayInit(1, POINTER_BYTES); - - pInfo->seqLoadData = false; - pInfo->pTransporter = pTransporter; - - pOperator->name = "ExchangeOperator"; - pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock); - pOperator->pTaskInfo = pTaskInfo; - - pOperator->fpSet = - createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, NULL, destroyExchangeOperatorInfo, NULL); - return pOperator; - -_error: - if (pInfo != NULL) { - doDestroyExchangeOperatorInfo(pInfo); - } - - taosMemoryFreeClear(pOperator); - pTaskInfo->code = code; - return NULL; -} - static int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t keyBufSize, const char* pKey); @@ -2960,7 +2369,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiN pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, NULL, destroyAggOperatorInfo, NULL); + createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, destroyAggOperatorInfo, NULL); if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) { STableScanInfo* pTableScanInfo = downstream->info; @@ -3024,33 +2433,6 @@ void destroyFillOperatorInfo(void* param) { taosMemoryFreeClear(param); } -void destroyExchangeOperatorInfo(void* param) { - SExchangeInfo* pExInfo = (SExchangeInfo*)param; - taosRemoveRef(exchangeObjRefPool, pExInfo->self); -} - -void freeSourceDataInfo(void* p) { - SSourceDataInfo* pInfo = (SSourceDataInfo*)p; - taosMemoryFreeClear(pInfo->pRsp); -} - -void doDestroyExchangeOperatorInfo(void* param) { - SExchangeInfo* pExInfo = (SExchangeInfo*)param; - - taosArrayDestroy(pExInfo->pSources); - taosArrayDestroyEx(pExInfo->pSourceDataInfo, freeSourceDataInfo); - - if (pExInfo->pResultBlockList != NULL) { - taosArrayDestroyEx(pExInfo->pResultBlockList, freeBlock); - pExInfo->pResultBlockList = NULL; - } - - blockDataDestroy(pExInfo->pDummyBlock); - - tsem_destroy(&pExInfo->ready); - taosMemoryFreeClear(param); -} - static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, SExprInfo* pNotFillExpr, int32_t numOfNotFillCols, SNodeListNode* pValNode, STimeWindow win, int32_t capacity, const char* id, SInterval* pInterval, int32_t fillType, int32_t order) { @@ -3190,7 +2572,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, NULL, destroyFillOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, destroyFillOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); return pOperator; diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 891eb6e7a4..cfcdd0ee64 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -446,7 +446,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, hashGroupbyAggregate, NULL, NULL, destroyGroupOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, hashGroupbyAggregate, NULL, destroyGroupOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -831,7 +831,7 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, NULL, destroyPartitionOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, destroyPartitionOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); return pOperator; @@ -1111,7 +1111,7 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamHashPartition, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamHashPartition, NULL, destroyStreamPartitionOperatorInfo, NULL); initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup); diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 45d76dce74..e7fe2bfc12 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -121,8 +121,7 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t pInfo->inputOrder = TSDB_ORDER_DESC; } - pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, NULL, destroyMergeJoinOperator, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, destroyMergeJoinOperator, NULL); code = appendDownstream(pOperator, pDownstream, numOfDownstream); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -372,13 +371,13 @@ static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) if (leftTs == rightTs) { mergeJoinJoinDownstreamTsRanges(pOperator, leftTs, pRes, &nrows); - } else if (asc && leftTs < rightTs || !asc && leftTs > rightTs) { + } else if ((asc && leftTs < rightTs) || (!asc && leftTs > rightTs)) { pJoinInfo->leftPos += 1; if (pJoinInfo->leftPos >= pJoinInfo->pLeft->info.rows) { continue; } - } else if (asc && leftTs > rightTs || !asc && leftTs < rightTs) { + } else if ((asc && leftTs > rightTs) || (!asc && leftTs < rightTs)) { pJoinInfo->rightPos += 1; if (pJoinInfo->rightPos >= pJoinInfo->pRight->info.rows) { continue; diff --git a/source/libs/executor/src/projectoperator.c b/source/libs/executor/src/projectoperator.c index b2865d15f0..58cc082653 100644 --- a/source/libs/executor/src/projectoperator.c +++ b/source/libs/executor/src/projectoperator.c @@ -104,7 +104,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, destroyProjectOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); @@ -406,8 +406,7 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doApplyIndefinitFunction, NULL, NULL, - destroyIndefinitOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doApplyIndefinitFunction, NULL, destroyIndefinitOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 6d13048123..903843bcc4 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -962,7 +962,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, } taosLRUCacheSetStrictCapacity(pInfo->metaCache.pTableMetaEntryCache, false); - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, destroyTableScanOperatorInfo, getTableScannerExecInfo); // for non-blocking operator, the open cost is always 0 @@ -993,7 +993,7 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL); return pOperator; } @@ -1155,8 +1155,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, NULL, destroyBlockDistScanOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL); return pOperator; _error: @@ -2372,7 +2371,7 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, destroyRawScanOperatorInfo, NULL); return pOperator; _end: @@ -2565,7 +2564,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys pOperator->pTaskInfo = pTaskInfo; __optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, nextFn, NULL, NULL, destroyStreamScanOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, NULL); return pOperator; @@ -4207,7 +4206,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock); pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, NULL); return pOperator; _error: @@ -4346,7 +4345,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi initResultSizeInfo(&pOperator->resultInfo, 4096); blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, NULL, destroyTagScanOperatorInfo, NULL); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, NULL); return pOperator; @@ -4861,7 +4860,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, destroyTableMergeScanOperatorInfo, getTableMergeScanExplainExecInfo); pOperator->cost.openCost = 0; return pOperator; diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index af87c4b2eb..700612e0ab 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -67,7 +67,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* // TODO dynamic set the available sort buffer pOperator->fpSet = - createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroySortOperatorInfo, getExplainExecInfo); + createOperatorFpSet(doOpenSortOperator, doSort, NULL, destroySortOperatorInfo, getExplainExecInfo); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -517,7 +517,7 @@ SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSort pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doGroupSort, NULL, NULL, destroyGroupSortOperatorInfo, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doGroupSort, NULL, destroyGroupSortOperatorInfo, getGroupSortExplainExecInfo); code = appendDownstream(pOperator, &downstream, 1); @@ -781,7 +781,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->fpSet = createOperatorFpSet(doOpenMultiwayMergeOperator, doMultiwayMerge, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(doOpenMultiwayMergeOperator, doMultiwayMerge, NULL, destroyMultiwayMergeOperatorInfo, getMultiwayMergeExplainExecInfo); code = appendDownstream(pOperator, downStreams, numStreams); diff --git a/source/libs/executor/src/tfill.c b/source/libs/executor/src/tfill.c index 9f06e639b3..a6ac28829c 100644 --- a/source/libs/executor/src/tfill.c +++ b/source/libs/executor/src/tfill.c @@ -1698,7 +1698,7 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doStreamFill, NULL, NULL, destroyStreamFillOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, doStreamFill, NULL, destroyStreamFillOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 7b6ed5b67d..6d8282ff5d 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1785,7 +1785,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPh pOperator->info = pInfo; pOperator->fpSet = - createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, NULL, destroyIntervalOperatorInfo, NULL); + createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, destroyIntervalOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -2565,7 +2565,7 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, NULL, destroyTimeSliceOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, destroyTimeSliceOperatorInfo, NULL); blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); @@ -2641,7 +2641,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi pOperator->info = pInfo; pOperator->fpSet = - createOperatorFpSet(openStateWindowAggOptr, doStateWindowAgg, NULL, NULL, destroyStateWindowOperatorInfo, NULL); + createOperatorFpSet(openStateWindowAggOptr, doStateWindowAgg, NULL, destroyStateWindowOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -2718,7 +2718,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW pOperator->info = pInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doSessionWindowAgg, NULL, NULL, destroySWindowOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, doSessionWindowAgg, NULL, destroySWindowOperatorInfo, NULL); pOperator->pTaskInfo = pTaskInfo; code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -3403,7 +3403,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, pOperator->info = pInfo; pOperator->fpSet = - createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo, NULL); + createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, destroyStreamFinalIntervalOperatorInfo, NULL); if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL) { initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup); } @@ -4196,7 +4196,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL); if (downstream) { initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType, @@ -4342,7 +4342,7 @@ SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR); blockDataEnsureCapacity(pInfo->pUpdateRes, 128); pOperator->name = "StreamSessionSemiAggOperator"; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL); } @@ -4713,7 +4713,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys pOperator->pTaskInfo = pTaskInfo; pOperator->info = pInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, NULL, destroyStreamStateOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL); initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType, pInfo->primaryTsIndex); code = appendDownstream(pOperator, &downstream, 1); @@ -4995,7 +4995,7 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, pOperator->info = miaInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, mergeAlignedIntervalAgg, NULL, NULL, destroyMAIOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, mergeAlignedIntervalAgg, NULL, destroyMAIOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -5307,7 +5307,7 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMerge pOperator->info = pMergeIntervalInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doMergeIntervalAgg, NULL, NULL, destroyMergeIntervalOperatorInfo, NULL); + createOperatorFpSet(operatorDummyOpenFn, doMergeIntervalAgg, NULL, destroyMergeIntervalOperatorInfo, NULL); code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -5540,7 +5540,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL, NULL, + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL, destroyStreamFinalIntervalOperatorInfo, NULL); initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup); diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index ad342edfa0..573a1ea31f 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -118,8 +118,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) pRetrieve->ekey = htobe64(pBlock->info.window.ekey); pRetrieve->version = htobe64(pBlock->info.version); - int32_t actualLen = 0; - blockEncode(pBlock, pRetrieve->data, &actualLen, numOfCols, false); + int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols); SStreamRetrieveReq req = { .streamId = pTask->streamId, @@ -200,8 +199,7 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock); pRetrieve->numOfCols = htonl(numOfCols); - int32_t actualLen = 0; - blockEncode(pBlock, pRetrieve->data, &actualLen, numOfCols, false); + int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols); actualLen += sizeof(SRetrieveTableRsp); ASSERT(actualLen <= dataStrLen); taosArrayPush(pReq->dataLen, &actualLen);