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

feat: ordered region and blocks

This commit is contained in:
shenglian zhou 2024-03-01 17:05:24 +08:00 committed by slzhou
parent 049f27bf3e
commit b6095261f2
3 changed files with 166 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/executor/inc/tsort.h
View File

@ -194,7 +194,7 @@ void tsortSetClosed(SSortHandle* pHandle);
void tsortSetSingleTableMerge(SSortHandle* pHandle); void tsortSetSingleTableMerge(SSortHandle* pHandle);
void tsortSetAbortCheckFn(SSortHandle* pHandle, bool (*checkFn)(void* param), void* param); void tsortSetAbortCheckFn(SSortHandle* pHandle, bool (*checkFn)(void* param), void* param);
int32_t tsortSetSortByRowId(SSortHandle* pHandle, int32_t extRowsPageSize, int32_t extRowsSize); int32_t tsortSetSortByRowId(SSortHandle* pHandle, int32_t extRowsSize);
void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHandle* pTupleHandle); void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHandle* pTupleHandle);
/** /**

2
source/libs/executor/src/scanoperator.c
View File

@ -4015,7 +4015,7 @@ int32_t startDurationForGroupTableMergeScan(SOperatorInfo* pOperator) {
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage, pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_BLOCK_TS_MERGE, pInfo->bufPageSize, numOfBufPage,
pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0); pInfo->pSortInputBlock, pTaskInfo->id.str, 0, 0, 0);
int32_t memSize = 512 * 1024 * 1024; int32_t memSize = 512 * 1024 * 1024;
code = tsortSetSortByRowId(pInfo->pSortHandle, pInfo->bufPageSize, memSize); code = tsortSetSortByRowId(pInfo->pSortHandle, memSize);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }

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

@ -45,23 +45,27 @@ typedef struct SSortMemPageEntry {
} SSortMemPageEntry; } SSortMemPageEntry;
typedef struct SSortMemFile { typedef struct SSortMemFileRegion {
int32_t pageSize; int64_t fileOffset;
int32_t cacheSize; int32_t regionSize;
char* writePageBuf;
int32_t startPageId;
int32_t numWritePages;
int32_t currPageId;
int32_t currPageOffset;
bool bDirty;
int32_t totalMemPages; int32_t bufRegOffset;
SSortMemPageEntry* pagesHead; int32_t bufLen;
SSortMemPageEntry* pagesTail; char* buf;
int32_t numMemPages; } SSortMemFileRegion;
SSHashObj* mActivePages;
typedef struct SSortMemFile {
char* writeBuf;
int32_t writeBufSize;
int64_t writeFileOffset;
int32_t currRegionId;
int32_t currRegionOffset;
bool bRegionDirty;
SArray* aFileRegions;
int32_t cacheSize;
int32_t blockSize;
FILE* pTdFile; FILE* pTdFile;
// TdFilePtr pTdFile; // TdFilePtr pTdFile;
@ -125,11 +129,8 @@ struct SSortHandle {
}; };
static int32_t destroySortMemFile(SSortHandle* pHandle); static int32_t destroySortMemFile(SSortHandle* pHandle);
static int32_t getPageFromExtMemFile(SSortHandle* pHandle, int32_t pageId, char** ppPage); static int32_t getRowBufFromExtMemFile(SSortHandle* pHandle, int32_t regionId, int32_t tupleOffset, int32_t rowLen,
static void setExtMemFilePageUnused(SSortMemFile* pMemFile, int32_t pageId); char** ppRow, bool* pFreeRow);
static int32_t saveDirtyPagesToExtRowsMemFile(SSortHandle* pHandle);
static int32_t freeExtRowMemFileWriteBuf(SSortHandle* pHandle);
void tsortSetSingleTableMerge(SSortHandle* pHandle) { void tsortSetSingleTableMerge(SSortHandle* pHandle) {
pHandle->singleTableMerge = true; pHandle->singleTableMerge = true;
} }
@ -915,14 +916,14 @@ static int32_t createPageBuf(SSortHandle* pHandle) {
void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHandle* pTupleHandle) { void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHandle* pTupleHandle) {
if (pHandle->bSortByRowId) { if (pHandle->bSortByRowId) {
int32_t pageId = *(int32_t*)tsortGetValue(pTupleHandle, 1); int32_t regionId = *(int32_t*)tsortGetValue(pTupleHandle, 1);
int32_t offset = *(int32_t*)tsortGetValue(pTupleHandle, 2); int32_t offset = *(int32_t*)tsortGetValue(pTupleHandle, 2);
int32_t length = *(int32_t*)tsortGetValue(pTupleHandle, 3);
char* page = NULL; char* buf = NULL;
getPageFromExtMemFile(pHandle, pageId, &page); bool bFreeRow = false;
getRowBufFromExtMemFile(pHandle, regionId, offset, length, &buf, &bFreeRow);
int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock); int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
char* buf = (char*)page + offset;
char* isNull = (char*)buf; char* isNull = (char*)buf;
char* pStart = (char*)buf + sizeof(int8_t) * numOfCols; char* pStart = (char*)buf + sizeof(int8_t) * numOfCols;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
@ -943,7 +944,9 @@ void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHa
colDataSetNULL(pColInfo, pBlock->info.rows); colDataSetNULL(pColInfo, pBlock->info.rows);
} }
} }
if (bFreeRow) {
taosMemoryFree(buf);
}
if (*(int32_t*)pStart != pStart - buf) { if (*(int32_t*)pStart != pStart - buf) {
qError("table merge scan row buf deserialization. length error %d != %d ", *(int32_t*)pStart, qError("table merge scan row buf deserialization. length error %d != %d ", *(int32_t*)pStart,
(int32_t)(pStart - buf)); (int32_t)(pStart - buf));
@ -953,9 +956,6 @@ void tsortAppendTupleToBlock(SSortHandle* pHandle, SSDataBlock* pBlock, STupleHa
pBlock->info.scanFlag = ((SDataBlockInfo*)tsortGetBlockInfo(pTupleHandle))->scanFlag; pBlock->info.scanFlag = ((SDataBlockInfo*)tsortGetBlockInfo(pTupleHandle))->scanFlag;
pBlock->info.rows += 1; pBlock->info.rows += 1;
if (offset + pHandle->extRowBytes >= pHandle->pExtRowsMemFile->pageSize) {
setExtMemFilePageUnused(pHandle->pExtRowsMemFile, pageId);
}
} else { } else {
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i); SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
@ -1020,102 +1020,60 @@ static int32_t blockRowToBuf(SSDataBlock* pBlock, int32_t rowIdx, char* buf) {
return (int32_t)(pStart - (char*)buf); return (int32_t)(pStart - (char*)buf);
} }
static int32_t getPageFromExtMemFile(SSortHandle* pHandle, int32_t pageId, char** ppPage) { static int32_t getRowBufFromExtMemFile(SSortHandle* pHandle, int32_t regionId, int32_t tupleOffset, int32_t rowLen,
char** ppRow, bool* pFreeRow) {
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile; SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
SSortMemPageEntry** ppPageEntry = tSimpleHashGet(pMemFile->mActivePages, &pageId, sizeof(pageId)); SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, regionId);
if (ppPageEntry) { if (pRegion->buf == NULL) {
*ppPage = (char*)((*ppPageEntry)->data); pRegion->bufRegOffset = 0;
pRegion->buf = taosMemoryMalloc(pMemFile->blockSize);
tsortSeekFile(pMemFile->pTdFile, pRegion->fileOffset, SEEK_SET);
int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize);
fread(pRegion->buf, readBytes, 1, pMemFile->pTdFile);
pRegion->bufLen = readBytes;
}
// TODO: ASSERT(pRegion->offset < tupleOffset);
if (pRegion->bufRegOffset + pRegion->bufLen >= tupleOffset + rowLen) {
*pFreeRow = false;
*ppRow = pRegion->buf + tupleOffset - pRegion->bufRegOffset;
} else { } else {
SSortMemPageEntry* pEntry = pMemFile->pagesHead->next; *ppRow = taosMemoryMalloc(rowLen);
bool freeEntryWhenError = false; int32_t szThisBlock = pRegion->bufLen - (tupleOffset - pRegion->bufRegOffset);
if (pEntry && !pEntry->active || pMemFile->numMemPages >= pMemFile->totalMemPages) { memcpy(*ppRow, pRegion->buf + tupleOffset - pRegion->bufRegOffset,
if (pEntry->active) { szThisBlock);
tSimpleHashRemove(pMemFile->mActivePages, &pEntry->pageId, sizeof(pEntry->pageId)); tsortSeekFile(pMemFile->pTdFile, pRegion->fileOffset + pRegion->bufRegOffset + pRegion->bufLen, SEEK_SET);
} int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize - (pRegion->bufRegOffset + pRegion->bufLen));
pEntry->prev->next = pEntry->next; fread(pRegion->buf, readBytes, 1, pMemFile->pTdFile);
pEntry->next->prev = pEntry->prev; memcpy(*ppRow + szThisBlock, pRegion->buf, rowLen - szThisBlock);
pEntry->active = false; *pFreeRow = true;
} else if (pMemFile->numMemPages < pMemFile->totalMemPages) { pRegion->bufRegOffset += pRegion->bufLen;
pEntry = taosMemoryCalloc(1, sizeof(SSortMemPageEntry)); pRegion->bufLen = readBytes;
pEntry->data = taosMemoryMalloc(pMemFile->pageSize);
freeEntryWhenError = true;
++pMemFile->numMemPages;
}
{
int ret = tsortSeekFile(pMemFile->pTdFile, ((int64_t)pageId) * pMemFile->pageSize, SEEK_SET);
if (ret == 0) {
ret = fread(pEntry->data, pMemFile->pageSize, 1, pMemFile->pTdFile);
}
if (ret != 1) {
terrno = TAOS_SYSTEM_ERROR(errno);
if (freeEntryWhenError) {
taosMemoryFreeClear(pEntry->data);
taosMemoryFreeClear(pEntry);
}
return terrno;
}
SSortMemPageEntry* tail = pMemFile->pagesTail;
tail->next = pEntry;
pEntry->next = NULL;
pEntry->prev = tail;
pEntry->active = true;
pMemFile->pagesTail = pEntry;
tSimpleHashPut(pMemFile->mActivePages, &pageId, sizeof(pageId), &pEntry, POINTER_BYTES);
*ppPage = pEntry->data;
}
} }
//TODO: free region memory
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void setExtMemFilePageUnused(SSortMemFile* pMemFile, int32_t pageId) {
SSortMemPageEntry** ppPageEntry = tSimpleHashGet(pMemFile->mActivePages, &pageId, sizeof(pageId));
SSortMemPageEntry* pEntry = *ppPageEntry;
if (pEntry == pMemFile->pagesTail) {
pMemFile->pagesTail = pEntry->prev;
}
pEntry->prev->next = pEntry->next;
pEntry->next->prev = pEntry->prev;
SSortMemPageEntry* first = pMemFile->pagesHead->next;
SSortMemPageEntry* head = pMemFile->pagesHead;
head->next = pEntry;
pEntry->next = first;
first->prev = pEntry;
pEntry->prev = head;
pEntry->active = false;
tSimpleHashRemove(pMemFile->mActivePages, &pageId, sizeof(pageId));
return;
}
static int32_t createSortMemFile(SSortHandle* pHandle) { static int32_t createSortMemFile(SSortHandle* pHandle) {
if (pHandle->pExtRowsMemFile != NULL) { if (pHandle->pExtRowsMemFile != NULL) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SSortMemFile* pMemFile = taosMemoryCalloc(1, sizeof(SSortMemFile)); SSortMemFile* pMemFile = taosMemoryCalloc(1, sizeof(SSortMemFile));
pMemFile->cacheSize = pHandle->extRowsMemSize;
taosGetTmpfilePath(tsTempDir, "sort-ext-mem", pMemFile->memFilePath); taosGetTmpfilePath(tsTempDir, "sort-ext-mem", pMemFile->memFilePath);
pMemFile->pTdFile = fopen(pMemFile->memFilePath, "w+"); pMemFile->pTdFile = fopen(pMemFile->memFilePath, "w+");
if (pMemFile->pTdFile == NULL) { if (pMemFile->pTdFile == NULL) {
taosMemoryFree(pMemFile); taosMemoryFree(pMemFile);
return TAOS_SYSTEM_ERROR(errno); return TAOS_SYSTEM_ERROR(errno);
} }
pMemFile->currPageId = -1; pMemFile->currRegionId = -1;
pMemFile->currPageOffset = -1; pMemFile->currRegionOffset = -1;
pMemFile->pageSize = pHandle->extRowsPageSize; pMemFile->writeBufSize = 64 * 1024 * 1024;
pMemFile->cacheSize = pHandle->extRowsMemSize; pMemFile->writeBuf = taosMemoryMalloc(pMemFile->writeBufSize);
pMemFile->numWritePages = pMemFile->cacheSize/pMemFile->pageSize; pMemFile->writeFileOffset = -1;
pMemFile->writePageBuf = taosMemoryMalloc(pMemFile->pageSize * pMemFile->numWritePages); pMemFile->bRegionDirty = false;
pMemFile->bDirty = false;
pMemFile->mActivePages = tSimpleHashInit(8192, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT)); pMemFile->aFileRegions = taosArrayInit(64, sizeof(SSortMemFileRegion));
pMemFile->pagesHead = taosMemoryCalloc(1, sizeof(SSortMemPageEntry));
pMemFile->pagesTail = pMemFile->pagesHead;
pMemFile->totalMemPages = pMemFile->cacheSize / pMemFile->pageSize;
pMemFile->numMemPages = 0;
pHandle->pExtRowsMemFile = pMemFile; pHandle->pExtRowsMemFile = pMemFile;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1125,20 +1083,15 @@ static int32_t destroySortMemFile(SSortHandle* pHandle) {
if (pHandle->pExtRowsMemFile == NULL) return TSDB_CODE_SUCCESS; if (pHandle->pExtRowsMemFile == NULL) return TSDB_CODE_SUCCESS;
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile; SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
SSortMemPageEntry* pEntry = pMemFile->pagesHead; for (int32_t i = 0; i < taosArrayGetSize(pMemFile->aFileRegions); ++i) {
while (pEntry != NULL) { SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, i);
if (pEntry->data) { taosMemoryFree(pRegion->buf);
taosMemoryFree(pEntry->data);
}
SSortMemPageEntry* pCurr = pEntry;
pEntry = pEntry->next;
taosMemoryFree(pCurr);
} }
tSimpleHashCleanup(pMemFile->mActivePages); taosArrayDestroy(pMemFile->aFileRegions);
pMemFile->mActivePages = NULL; pMemFile->aFileRegions = NULL;
taosMemoryFree(pMemFile->writePageBuf); taosMemoryFree(pMemFile->writeBuf);
pMemFile->writePageBuf = NULL; pMemFile->writeBuf = NULL;
fclose(pMemFile->pTdFile); fclose(pMemFile->pTdFile);
pMemFile->pTdFile = NULL; pMemFile->pTdFile = NULL;
@ -1148,68 +1101,89 @@ static int32_t destroySortMemFile(SSortHandle* pHandle) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t saveBlockRowToExtRowsMemFile(SSortHandle* pHandle, SSDataBlock* pBlock, int32_t rowIdx, int32_t* pPageId, int32_t* pOffset, int32_t* pLength) { static int32_t tsortOpenRegion(SSortHandle* pHandle) {
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile; SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
if (pMemFile->currPageId == -1) { if (pMemFile->currRegionId == -1) {
pMemFile->currPageId = 0; SSortMemFileRegion region = {0};
pMemFile->currPageOffset = 0; region.fileOffset = 0;
pMemFile->startPageId = 0; region.bufRegOffset = 0;
taosArrayPush(pMemFile->aFileRegions, &region);
pMemFile->currRegionId = 0;
pMemFile->currRegionOffset = 0;
pMemFile->writeFileOffset = 0;
} else { } else {
if (pMemFile->currPageOffset + pHandle->extRowBytes >= pMemFile->pageSize) { SSortMemFileRegion regionNew = {0};
SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, pMemFile->currRegionId);
regionNew.fileOffset = pRegion->fileOffset + pRegion->regionSize;
regionNew.bufRegOffset = 0;
taosArrayPush(pMemFile->aFileRegions, &regionNew);
++pMemFile->currRegionId;
pMemFile->currRegionOffset = 0;
pMemFile->writeFileOffset = regionNew.fileOffset;
}
return TSDB_CODE_SUCCESS;
}
++pMemFile->currPageId; static int32_t tsortCloseRegion(SSortHandle* pHandle) {
pMemFile->currPageOffset = 0; SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, pMemFile->currRegionId);
pRegion->regionSize = pMemFile->currRegionOffset;
int32_t writeBytes = pRegion->regionSize - (pMemFile->writeFileOffset - pRegion->fileOffset);
if (writeBytes > 0) {
int ret = tsortSeekFile(pMemFile->pTdFile, pMemFile->writeFileOffset, SEEK_SET);
if (ret == 0) {
ret = fwrite(pMemFile->writeBuf, writeBytes, 1, pMemFile->pTdFile);
}
if (ret != 1) {
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
}
pMemFile->bRegionDirty = false;
}
return TSDB_CODE_SUCCESS;
}
if (pMemFile->currPageId - pMemFile->startPageId >= pMemFile->numWritePages) { static int32_t tsortFinalizeRegions(SSortHandle* pHandle) {
int ret = tsortSeekFile(pMemFile->pTdFile, ((int64_t)pMemFile->startPageId) * pMemFile->pageSize, SEEK_SET); SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
if (ret == 0) { size_t numRegions = taosArrayGetSize(pMemFile->aFileRegions);
ret = fwrite(pMemFile->writePageBuf, pMemFile->pageSize * pMemFile->numWritePages, 1, pMemFile->pTdFile); ASSERT(numRegions == (pMemFile->currRegionId + 1));
} if (numRegions == 0) return TSDB_CODE_SUCCESS;
if (ret != 1) { int32_t blockReadBytes = (pMemFile->cacheSize / numRegions + 4095) & ~4095;
terrno = TAOS_SYSTEM_ERROR(errno); pMemFile->blockSize = blockReadBytes;
return terrno;
} for (int32_t i = 0; i < numRegions; ++i) {
pMemFile->startPageId = pMemFile->currPageId; SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, i);
pRegion->bufRegOffset = 0;
}
taosMemoryFree(pMemFile->writeBuf);
pMemFile->writeBuf = NULL;
return TSDB_CODE_SUCCESS;
}
static int32_t saveBlockRowToExtRowsMemFile(SSortHandle* pHandle, SSDataBlock* pBlock, int32_t rowIdx, int32_t* pRegionId, int32_t* pOffset, int32_t* pLength) {
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
SSortMemFileRegion* pRegion = taosArrayGet(pMemFile->aFileRegions, pMemFile->currRegionId);
{
if (pMemFile->currRegionOffset + pHandle->extRowBytes >= pMemFile->writeBufSize) {
int ret = tsortSeekFile(pMemFile->pTdFile, pMemFile->writeFileOffset, SEEK_SET);
int32_t writeBytes = pMemFile->currRegionOffset - (pMemFile->writeFileOffset - pRegion->fileOffset);
if (ret == 0) {
ret = fwrite(pMemFile->writeBuf, writeBytes, 1, pMemFile->pTdFile);
} }
if (ret != 1) {
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
}
pMemFile->writeFileOffset = pRegion->fileOffset + pMemFile->currRegionOffset;
} }
} }
*pRegionId = pMemFile->currRegionId;
*pPageId = pMemFile->currPageId; *pOffset = pMemFile->currRegionOffset;
*pOffset = pMemFile->currPageOffset; int32_t writeBufOffset = pMemFile->currRegionOffset - (pMemFile->writeFileOffset - pRegion->fileOffset);
int32_t offsetPages = (pMemFile->currPageId - pMemFile->startPageId) * pMemFile->pageSize; int32_t blockLen = blockRowToBuf(pBlock, rowIdx, pMemFile->writeBuf + writeBufOffset);
int32_t blockLen = blockRowToBuf(pBlock, rowIdx,
pMemFile->writePageBuf + offsetPages + pMemFile->currPageOffset);
*pLength = blockLen; *pLength = blockLen;
pMemFile->currPageOffset += blockLen; pMemFile->currRegionOffset += blockLen;
pMemFile->bDirty = true; pMemFile->bRegionDirty = true;
return TSDB_CODE_SUCCESS;
}
static int32_t saveDirtyPagesToExtRowsMemFile(SSortHandle* pHandle) {
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
if (!pMemFile->bDirty) {
return TSDB_CODE_SUCCESS;
}
int ret = tsortSeekFile(pMemFile->pTdFile, ((int64_t)pMemFile->startPageId) * pMemFile->pageSize, SEEK_SET);
int32_t numWriteBytes = pMemFile->pageSize * (pMemFile->currPageId - pMemFile->startPageId + 1);
if (ret == 0) {
ret = fwrite(pMemFile->writePageBuf, numWriteBytes, 1, pMemFile->pTdFile);
}
if (ret != 1) {
terrno = TAOS_SYSTEM_ERROR(errno);
return terrno;
}
pMemFile->bDirty = false;
return TSDB_CODE_SUCCESS;
}
static int32_t freeExtRowMemFileWriteBuf(SSortHandle* pHandle) {
SSortMemFile* pMemFile = pHandle->pExtRowsMemFile;
if (pMemFile == NULL) return TSDB_CODE_SUCCESS;
taosMemoryFree(pMemFile->writePageBuf);
pMemFile->writePageBuf = NULL;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1225,12 +1199,15 @@ static void appendToRowIndexDataBlock(SSortHandle* pHandle, SSDataBlock* pSource
char* pData = colDataGetData(pSrcTsCol, *rowIndex); char* pData = colDataGetData(pSrcTsCol, *rowIndex);
colDataSetVal(pTsCol, pBlock->info.rows, pData, false); colDataSetVal(pTsCol, pBlock->info.rows, pData, false);
SColumnInfoData* pPageIdCol = taosArrayGet(pBlock->pDataBlock, 1); SColumnInfoData* pRegionIdCol = taosArrayGet(pBlock->pDataBlock, 1);
colDataSetInt32(pPageIdCol, pBlock->info.rows, &pageId); colDataSetInt32(pRegionIdCol, pBlock->info.rows, &pageId);
SColumnInfoData* pOffsetCol = taosArrayGet(pBlock->pDataBlock, 2); SColumnInfoData* pOffsetCol = taosArrayGet(pBlock->pDataBlock, 2);
colDataSetInt32(pOffsetCol, pBlock->info.rows, &offset); colDataSetInt32(pOffsetCol, pBlock->info.rows, &offset);
SColumnInfoData* pLengthCol = taosArrayGet(pBlock->pDataBlock, 3);
colDataSetInt32(pLengthCol, pBlock->info.rows, &length);
pBlock->info.rows += 1; pBlock->info.rows += 1;
*rowIndex += 1; *rowIndex += 1;
} }
@ -1240,10 +1217,12 @@ static void initRowIdSort(SSortHandle* pHandle) {
SSDataBlock* pSortInput = createDataBlock(); SSDataBlock* pSortInput = createDataBlock();
SColumnInfoData tsCol = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, 8, 1); SColumnInfoData tsCol = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, 8, 1);
blockDataAppendColInfo(pSortInput, &tsCol); blockDataAppendColInfo(pSortInput, &tsCol);
SColumnInfoData pageIdCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 2); SColumnInfoData regionIdCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 2);
blockDataAppendColInfo(pSortInput, &pageIdCol); blockDataAppendColInfo(pSortInput, &regionIdCol);
SColumnInfoData offsetCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 3); SColumnInfoData offsetCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 3);
blockDataAppendColInfo(pSortInput, &offsetCol); blockDataAppendColInfo(pSortInput, &offsetCol);
SColumnInfoData lengthCol = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 4);
blockDataAppendColInfo(pSortInput, &lengthCol);
blockDataDestroy(pHandle->pDataBlock); blockDataDestroy(pHandle->pDataBlock);
pHandle->pDataBlock = pSortInput; pHandle->pDataBlock = pSortInput;
@ -1267,9 +1246,8 @@ static void initRowIdSort(SSortHandle* pHandle) {
return; return;
} }
int32_t tsortSetSortByRowId(SSortHandle* pHandle, int32_t extRowsPageSize, int32_t extRowsMemSize) { int32_t tsortSetSortByRowId(SSortHandle* pHandle, int32_t extRowsMemSize) {
pHandle->extRowBytes = blockDataGetRowSize(pHandle->pDataBlock) + taosArrayGetSize(pHandle->pDataBlock->pDataBlock) + sizeof(int32_t); pHandle->extRowBytes = blockDataGetRowSize(pHandle->pDataBlock) + taosArrayGetSize(pHandle->pDataBlock->pDataBlock) + sizeof(int32_t);
pHandle->extRowsPageSize = extRowsPageSize;
pHandle->extRowsMemSize = extRowsMemSize; pHandle->extRowsMemSize = extRowsMemSize;
SBlockOrderInfo* pOrder = taosArrayGet(pHandle->pSortInfo, 0); SBlockOrderInfo* pOrder = taosArrayGet(pHandle->pSortInfo, 0);
pHandle->extRowsOrderInfo = *pOrder; pHandle->extRowsOrderInfo = *pOrder;
@ -1552,7 +1530,6 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
while (1) { while (1) {
SSDataBlock* pBlk = pHandle->fetchfp(pSrc->param); SSDataBlock* pBlk = pHandle->fetchfp(pSrc->param);
int64_t p = taosGetTimestampUs();
bool bExtractedBlock = false; bool bExtractedBlock = false;
bool bSkipBlock = false; bool bSkipBlock = false;
if (pBlk != NULL && pHandle->mergeLimit > 0) { if (pBlk != NULL && pHandle->mergeLimit > 0) {
@ -1594,6 +1571,9 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
tSimpleHashClear(mUidBlk); tSimpleHashClear(mUidBlk);
int64_t p = taosGetTimestampUs(); int64_t p = taosGetTimestampUs();
if (pHandle->bSortByRowId) {
tsortOpenRegion(pHandle);
}
code = sortBlocksToExtSource(pHandle, aBlkSort, aExtSrc); code = sortBlocksToExtSource(pHandle, aBlkSort, aExtSrc);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -1603,7 +1583,9 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
taosArrayClear(aBlkSort); taosArrayClear(aBlkSort);
break; break;
} }
if (pHandle->bSortByRowId) {
tsortCloseRegion(pHandle);
}
int64_t el = taosGetTimestampUs() - p; int64_t el = taosGetTimestampUs() - p;
pHandle->sortElapsed += el; pHandle->sortElapsed += el;
@ -1641,8 +1623,7 @@ static int32_t createBlocksMergeSortInitialSources(SSortHandle* pHandle) {
taosArrayDestroy(aExtSrc); taosArrayDestroy(aExtSrc);
tSimpleHashCleanup(mTableNumRows); tSimpleHashCleanup(mTableNumRows);
if (pHandle->bSortByRowId) { if (pHandle->bSortByRowId) {
saveDirtyPagesToExtRowsMemFile(pHandle); tsortFinalizeRegions(pHandle);
freeExtRowMemFileWriteBuf(pHandle);
} }
pHandle->type = SORT_SINGLESOURCE_SORT; pHandle->type = SORT_SINGLESOURCE_SORT;
return code; return code;