diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 16780a9402ffbbff90bfb1624a4727c58edad94a..90dfbd5599cc609b8186df58914335a10a355bff 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -55,7 +55,6 @@ typedef struct STableMergeScanSortSourceParam { int32_t readerIdx; uint64_t uid; SSDataBlock* inputBlock; - bool multiReader; STsdbReader* dataReader; } STableMergeScanSortSourceParam; @@ -2658,8 +2657,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) { int64_t st = taosGetTimestampUs(); void* p = tableListGetInfo(pInfo->base.pTableListInfo, readIdx + pInfo->tableStartIndex); SReadHandle* pHandle = &pInfo->base.readHandle; - - if (NULL == source->dataReader || !source->multiReader) { + if (NULL == source->dataReader) { code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, (void**)&source->dataReader, GET_TASKID(pTaskInfo), false, NULL); if (code != 0) { T_LONG_JMP(pTaskInfo->env, code); @@ -2723,19 +2721,14 @@ static SSDataBlock* getTableDataBlockImpl(void* param) { pInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0; qTrace("tsdb/read-table-data: %p, close reader", reader); - if (!source->multiReader) { - pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader); - source->dataReader = NULL; - } pInfo->base.dataReader = NULL; return pBlock; } - if (!source->multiReader) { - pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader); - source->dataReader = NULL; - } + pAPI->tsdReader.tsdReaderClose(source->dataReader); + source->dataReader = NULL; pInfo->base.dataReader = NULL; + return NULL; } @@ -2791,7 +2784,8 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) { // todo the total available buffer should be determined by total capacity of buffer of this task. // the additional one is reserved for merge result - pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1); + // pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1); + pInfo->sortBufSize = pInfo->bufPageSize * (256 + 1); int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, pInfo->pSortInputBlock, pTaskInfo->id.str); @@ -2806,7 +2800,6 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) { STableMergeScanSortSourceParam param = {0}; param.readerIdx = i; param.pOperator = pOperator; - param.multiReader = (numOfTable <= MULTI_READER_MAX_TABLE_NUM) ? true : false; param.inputBlock = createOneDataBlock(pInfo->pResBlock, false); blockDataEnsureCapacity(param.inputBlock, pOperator->resultInfo.capacity); diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index 783597df6705af50017312095ede318707a2dd01..3033441aadf5b94c8cfb8f3ce29ac0a09b1b9907 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -101,7 +101,11 @@ static int32_t sortComparCleanup(SMsortComparParam* cmpParam) { for (int32_t i = 0; i < cmpParam->numOfSources; ++i) { SSortSource* pSource = cmpParam->pSources[i]; blockDataDestroy(pSource->src.pBlock); + if (pSource->pageIdList) { + taosArrayDestroy(pSource->pageIdList); + } taosMemoryFreeClear(pSource); + cmpParam->pSources[i] = NULL; } cmpParam->numOfSources = 0; @@ -123,9 +127,11 @@ void tsortClearOrderdSource(SArray* pOrderedSource, int64_t *fetchUs, int64_t *f // release pageIdList if ((*pSource)->pageIdList) { taosArrayDestroy((*pSource)->pageIdList); + (*pSource)->pageIdList = NULL; } if ((*pSource)->param && !(*pSource)->onlyRef) { taosMemoryFree((*pSource)->param); + (*pSource)->param = NULL; } if (!(*pSource)->onlyRef && (*pSource)->src.pBlock) {