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

enhance: merge blocks of the same table before sort

This commit is contained in:
slzhou 2023-07-20 09:28:19 +08:00
parent d1a1976aa4
commit 30129f64da
1 changed files with 16 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
source/libs/executor/src/tsort.c
View File

@ -24,6 +24,7 @@
#include "tpagedbuf.h" #include "tpagedbuf.h"
#include "tsort.h" #include "tsort.h"
#include "tutil.h" #include "tutil.h"
#include "tsimplehash.h"
struct STupleHandle { struct STupleHandle {
SSDataBlock* pBlock; SSDataBlock* pBlock;
@ -975,15 +976,27 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
int32_t szSort = 0; int32_t szSort = 0;
SArray* aBlkSort = taosArrayInit(8, POINTER_BYTES); SArray* aBlkSort = taosArrayInit(8, POINTER_BYTES);
SSHashObj* mUidBlk = tSimpleHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT));
while (1) { while (1) {
SSDataBlock* pBlk = pHandle->fetchfp(pSrc->param); SSDataBlock* pBlk = pHandle->fetchfp(pSrc->param);
if (pBlk != NULL) { if (pBlk != NULL) {
szSort += blockDataGetSize(pBlk); szSort += blockDataGetSize(pBlk);
SSDataBlock* blk = createOneDataBlock(pBlk, true);
taosArrayPush(aBlkSort, &blk); void* ppBlk = tSimpleHashGet(mUidBlk, &pBlk->info.id.uid, sizeof(pBlk->info.id.uid));
if (ppBlk != NULL) {
SSDataBlock* tBlk = *(SSDataBlock**)(ppBlk);
blockDataMerge(tBlk, pBlk);
} else {
SSDataBlock* tBlk = createOneDataBlock(pBlk, true);
tSimpleHashPut(mUidBlk, &pBlk->info.id.uid, sizeof(pBlk->info.id.uid), &tBlk, POINTER_BYTES);
taosArrayPush(aBlkSort, &tBlk);
}
} }
if ((pBlk != NULL && szSort > maxBufSize) || (pBlk == NULL && szSort > 0)) { if ((pBlk != NULL && szSort > maxBufSize) || (pBlk == NULL && szSort > 0)) {
tSimpleHashClear(mUidBlk);
int64_t p = taosGetTimestampUs(); int64_t p = taosGetTimestampUs();
sortBlocksToExtSource(pHandle, aBlkSort, pOrder, aExtSrc); sortBlocksToExtSource(pHandle, aBlkSort, pOrder, aExtSrc);
int64_t el = taosGetTimestampUs() - p; int64_t el = taosGetTimestampUs() - p;
@ -1000,7 +1013,7 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
break; break;
}; };
} }
tSimpleHashCleanup(mUidBlk);
taosArrayDestroy(aBlkSort); taosArrayDestroy(aBlkSort);
tsortClearOrderdSource(pHandle->pOrderedSource, NULL, NULL); tsortClearOrderdSource(pHandle->pOrderedSource, NULL, NULL);
taosArrayAddAll(pHandle->pOrderedSource, aExtSrc); taosArrayAddAll(pHandle->pOrderedSource, aExtSrc);