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

feat: use tdfile api

This commit is contained in:
slzhou 2024-03-07 09:52:56 +08:00
parent 4f8ff5b1a6
commit ffc8c8d148
1 changed files with 36 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -67,8 +67,7 @@ typedef struct SSortMemFile {
int32_t cacheSize; int32_t cacheSize;
int32_t blockSize; int32_t blockSize;
FILE* pTdFile; TdFilePtr pTdFile;
// TdFilePtr pTdFile;
char memFilePath[PATH_MAX]; char memFilePath[PATH_MAX];
} SSortMemFile; } SSortMemFile;
@ -1027,9 +1026,17 @@ static int32_t getRowBufFromExtMemFile(SSortHandle* pHandle, int32_t regionId, i
if (pRegion->buf == NULL) { if (pRegion->buf == NULL) {
pRegion->bufRegOffset = 0; pRegion->bufRegOffset = 0;
pRegion->buf = taosMemoryMalloc(pMemFile->blockSize); pRegion->buf = taosMemoryMalloc(pMemFile->blockSize);
tsortSeekFile(pMemFile->pTdFile, pRegion->fileOffset, SEEK_SET); if (pRegion->buf == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int64_t ret = taosLSeekFile(pMemFile->pTdFile, pRegion->fileOffset, SEEK_SET);
int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize); int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize);
fread(pRegion->buf, readBytes, 1, pMemFile->pTdFile); if (ret >= 0) {
ret = taosReadFile(pMemFile->pTdFile, pRegion->buf, readBytes);
}
if (ret != readBytes) {
return TAOS_SYSTEM_ERROR(errno);
}
pRegion->bufLen = readBytes; pRegion->bufLen = readBytes;
} }
// TODO: ASSERT(pRegion->offset < tupleOffset); // TODO: ASSERT(pRegion->offset < tupleOffset);
@ -1037,17 +1044,25 @@ static int32_t getRowBufFromExtMemFile(SSortHandle* pHandle, int32_t regionId, i
*pFreeRow = false; *pFreeRow = false;
*ppRow = pRegion->buf + tupleOffset - pRegion->bufRegOffset; *ppRow = pRegion->buf + tupleOffset - pRegion->bufRegOffset;
} else { } else {
*ppRow = taosMemoryMalloc(rowLen); *ppRow = taosMemoryMalloc(rowLen);
int32_t szThisBlock = pRegion->bufLen - (tupleOffset - pRegion->bufRegOffset); if (*ppRow == NULL) {
memcpy(*ppRow, pRegion->buf + tupleOffset - pRegion->bufRegOffset, return TSDB_CODE_OUT_OF_MEMORY;
szThisBlock); }
tsortSeekFile(pMemFile->pTdFile, pRegion->fileOffset + pRegion->bufRegOffset + pRegion->bufLen, SEEK_SET); int32_t szThisBlock = pRegion->bufLen - (tupleOffset - pRegion->bufRegOffset);
int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize - (pRegion->bufRegOffset + pRegion->bufLen)); memcpy(*ppRow, pRegion->buf + tupleOffset - pRegion->bufRegOffset,
fread(pRegion->buf, readBytes, 1, pMemFile->pTdFile); szThisBlock);
memcpy(*ppRow + szThisBlock, pRegion->buf, rowLen - szThisBlock); int64_t ret = taosLSeekFile(pMemFile->pTdFile, pRegion->fileOffset + pRegion->bufRegOffset + pRegion->bufLen, SEEK_SET);
*pFreeRow = true; int32_t readBytes = TMIN(pMemFile->blockSize, pRegion->regionSize - (pRegion->bufRegOffset + pRegion->bufLen));
pRegion->bufRegOffset += pRegion->bufLen; if (ret >= 0) {
pRegion->bufLen = readBytes; ret = taosReadFile(pMemFile->pTdFile, pRegion->buf, readBytes);
}
if (ret != readBytes) {
return TAOS_SYSTEM_ERROR(errno);
}
memcpy(*ppRow + szThisBlock, pRegion->buf, rowLen - szThisBlock);
*pFreeRow = true;
pRegion->bufRegOffset += pRegion->bufLen;
pRegion->bufLen = readBytes;
} }
//TODO: free region memory //TODO: free region memory
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1060,7 +1075,7 @@ static int32_t createSortMemFile(SSortHandle* pHandle) {
SSortMemFile* pMemFile = taosMemoryCalloc(1, sizeof(SSortMemFile)); SSortMemFile* pMemFile = taosMemoryCalloc(1, sizeof(SSortMemFile));
pMemFile->cacheSize = pHandle->extRowsMemSize; 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 = taosOpenFile(pMemFile->memFilePath, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);;
if (pMemFile->pTdFile == NULL) { if (pMemFile->pTdFile == NULL) {
taosMemoryFree(pMemFile); taosMemoryFree(pMemFile);
return TAOS_SYSTEM_ERROR(errno); return TAOS_SYSTEM_ERROR(errno);
@ -1093,8 +1108,7 @@ static int32_t destroySortMemFile(SSortHandle* pHandle) {
taosMemoryFree(pMemFile->writeBuf); taosMemoryFree(pMemFile->writeBuf);
pMemFile->writeBuf = NULL; pMemFile->writeBuf = NULL;
fclose(pMemFile->pTdFile); taosCloseFile(&pMemFile->pTdFile);
pMemFile->pTdFile = NULL;
taosRemoveFile(pMemFile->memFilePath); taosRemoveFile(pMemFile->memFilePath);
taosMemoryFree(pMemFile); taosMemoryFree(pMemFile);
pHandle->pExtRowsMemFile = NULL; pHandle->pExtRowsMemFile = NULL;
@ -1130,8 +1144,8 @@ static int32_t tsortCloseRegion(SSortHandle* pHandle) {
pRegion->regionSize = pMemFile->currRegionOffset; pRegion->regionSize = pMemFile->currRegionOffset;
int32_t writeBytes = pRegion->regionSize - (pMemFile->writeFileOffset - pRegion->fileOffset); int32_t writeBytes = pRegion->regionSize - (pMemFile->writeFileOffset - pRegion->fileOffset);
if (writeBytes > 0) { if (writeBytes > 0) {
int ret = fwrite(pMemFile->writeBuf, writeBytes, 1, pMemFile->pTdFile); int64_t ret = taosWriteFile(pMemFile->pTdFile, pMemFile->writeBuf, writeBytes);
if (ret != 1) { if (ret != writeBytes) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return terrno; return terrno;
} }
@ -1163,8 +1177,8 @@ static int32_t saveBlockRowToExtRowsMemFile(SSortHandle* pHandle, SSDataBlock* p
{ {
if (pMemFile->currRegionOffset + pHandle->extRowBytes >= pMemFile->writeBufSize) { if (pMemFile->currRegionOffset + pHandle->extRowBytes >= pMemFile->writeBufSize) {
int32_t writeBytes = pMemFile->currRegionOffset - (pMemFile->writeFileOffset - pRegion->fileOffset); int32_t writeBytes = pMemFile->currRegionOffset - (pMemFile->writeFileOffset - pRegion->fileOffset);
int ret = fwrite(pMemFile->writeBuf, writeBytes, 1, pMemFile->pTdFile); int64_t ret = taosWriteFile(pMemFile->pTdFile, pMemFile->writeBuf, writeBytes);
if (ret != 1) { if (ret != writeBytes) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return terrno; return terrno;
} }