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

tsdb/read-file-page: new arg for hint size

This commit is contained in:
Minglei Jin 2023-11-23 15:37:26 +08:00
parent 023821a80b
commit 08e5525012
3 changed files with 41 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
source/dnode/vnode/src/tsdb/tsdbDataFileRW.c
View File

@ -45,7 +45,7 @@ static int32_t tsdbDataFileReadHeadFooter(SDataFileReader *reader) {
int32_t ftype = TSDB_FTYPE_HEAD; int32_t ftype = TSDB_FTYPE_HEAD;
if (reader->fd[ftype]) { if (reader->fd[ftype]) {
code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(SHeadFooter), code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(SHeadFooter),
(uint8_t *)reader->headFooter, sizeof(SHeadFooter)); (uint8_t *)reader->headFooter, sizeof(SHeadFooter), 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
@ -67,7 +67,7 @@ static int32_t tsdbDataFileReadTombFooter(SDataFileReader *reader) {
int32_t ftype = TSDB_FTYPE_TOMB; int32_t ftype = TSDB_FTYPE_TOMB;
if (reader->fd[ftype]) { if (reader->fd[ftype]) {
code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(STombFooter), code = tsdbReadFile(reader->fd[ftype], reader->config->files[ftype].file.size - sizeof(STombFooter),
(uint8_t *)reader->tombFooter, sizeof(STombFooter)); (uint8_t *)reader->tombFooter, sizeof(STombFooter), 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
reader->ctx->tombFooterLoaded = true; reader->ctx->tombFooterLoaded = true;
@ -161,7 +161,7 @@ int32_t tsdbDataFileReadBrinBlk(SDataFileReader *reader, const TBrinBlkArray **b
} }
code = tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], reader->headFooter->brinBlkPtr->offset, data, code = tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], reader->headFooter->brinBlkPtr->offset, data,
reader->headFooter->brinBlkPtr->size); reader->headFooter->brinBlkPtr->size, 0);
if (code) { if (code) {
taosMemoryFree(data); taosMemoryFree(data);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
@ -191,7 +191,8 @@ int32_t tsdbDataFileReadBrinBlock(SDataFileReader *reader, const SBrinBlk *brinB
code = tRealloc(&reader->config->bufArr[0], brinBlk->dp->size); code = tRealloc(&reader->config->bufArr[0], brinBlk->dp->size);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], brinBlk->dp->offset, reader->config->bufArr[0], brinBlk->dp->size); code =
tsdbReadFile(reader->fd[TSDB_FTYPE_HEAD], brinBlk->dp->offset, reader->config->bufArr[0], brinBlk->dp->size, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
int32_t size = 0; int32_t size = 0;
@ -232,7 +233,8 @@ int32_t tsdbDataFileReadBlockData(SDataFileReader *reader, const SBrinRecord *re
code = tRealloc(&reader->config->bufArr[0], record->blockSize); code = tRealloc(&reader->config->bufArr[0], record->blockSize);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockSize); code =
tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockSize, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tDecmprBlockData(reader->config->bufArr[0], record->blockSize, bData, &reader->config->bufArr[1]); code = tDecmprBlockData(reader->config->bufArr[0], record->blockSize, bData, &reader->config->bufArr[1]);
@ -257,8 +259,8 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe
code = tRealloc(&reader->config->bufArr[0], record->blockKeySize); code = tRealloc(&reader->config->bufArr[0], record->blockKeySize);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockKeySize,
tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset, reader->config->bufArr[0], record->blockKeySize); 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
// hdr // hdr
@ -296,7 +298,7 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset + record->blockKeySize, code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], record->blockOffset + record->blockKeySize,
reader->config->bufArr[0], hdr->szBlkCol); reader->config->bufArr[0], hdr->szBlkCol, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
@ -374,7 +376,7 @@ int32_t tsdbDataFileReadBlockDataByColumn(SDataFileReader *reader, const SBrinRe
code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA], code = tsdbReadFile(reader->fd[TSDB_FTYPE_DATA],
record->blockOffset + record->blockKeySize + hdr->szBlkCol + blockCol->offset, record->blockOffset + record->blockKeySize + hdr->szBlkCol + blockCol->offset,
reader->config->bufArr[1], size1); reader->config->bufArr[1], size1, szHint);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbDecmprColData(reader->config->bufArr[1], blockCol, hdr->cmprAlg, hdr->nRow, colData, code = tsdbDecmprColData(reader->config->bufArr[1], blockCol, hdr->cmprAlg, hdr->nRow, colData,
@ -402,7 +404,7 @@ int32_t tsdbDataFileReadBlockSma(SDataFileReader *reader, const SBrinRecord *rec
code = tRealloc(&reader->config->bufArr[0], record->smaSize); code = tRealloc(&reader->config->bufArr[0], record->smaSize);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbReadFile(reader->fd[TSDB_FTYPE_SMA], record->smaOffset, reader->config->bufArr[0], record->smaSize); code = tsdbReadFile(reader->fd[TSDB_FTYPE_SMA], record->smaOffset, reader->config->bufArr[0], record->smaSize, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
// decode sma data // decode sma data
@ -441,7 +443,7 @@ int32_t tsdbDataFileReadTombBlk(SDataFileReader *reader, const TTombBlkArray **t
} }
code = tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], reader->tombFooter->tombBlkPtr->offset, data, code = tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], reader->tombFooter->tombBlkPtr->offset, data,
reader->tombFooter->tombBlkPtr->size); reader->tombFooter->tombBlkPtr->size, 0);
if (code) { if (code) {
taosMemoryFree(data); taosMemoryFree(data);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
@ -471,7 +473,8 @@ int32_t tsdbDataFileReadTombBlock(SDataFileReader *reader, const STombBlk *tombB
code = tRealloc(&reader->config->bufArr[0], tombBlk->dp->size); code = tRealloc(&reader->config->bufArr[0], tombBlk->dp->size);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size); code =
tsdbReadFile(reader->fd[TSDB_FTYPE_TOMB], tombBlk->dp->offset, reader->config->bufArr[0], tombBlk->dp->size, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
int32_t size = 0; int32_t size = 0;

2
source/dnode/vnode/src/tsdb/tsdbDef.h
View File

@ -34,7 +34,7 @@ typedef struct SFDataPtr {
extern int32_t tsdbOpenFile(const char *path, STsdb *pTsdb, int32_t flag, STsdbFD **ppFD); extern int32_t tsdbOpenFile(const char *path, STsdb *pTsdb, int32_t flag, STsdbFD **ppFD);
extern void tsdbCloseFile(STsdbFD **ppFD); extern void tsdbCloseFile(STsdbFD **ppFD);
extern int32_t tsdbWriteFile(STsdbFD *pFD, int64_t offset, const uint8_t *pBuf, int64_t size); extern int32_t tsdbWriteFile(STsdbFD *pFD, int64_t offset, const uint8_t *pBuf, int64_t size);
extern int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size); extern int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint);
extern int32_t tsdbFsyncFile(STsdbFD *pFD); extern int32_t tsdbFsyncFile(STsdbFD *pFD);
#ifdef __cplusplus #ifdef __cplusplus

36
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
View File

@ -283,7 +283,7 @@ _exit:
return code; return code;
} }
static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) { static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint) {
int32_t code = 0; int32_t code = 0;
int64_t n = 0; int64_t n = 0;
int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage); int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage);
@ -339,6 +339,11 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64
uint8_t *pBlock = NULL; uint8_t *pBlock = NULL;
int64_t retrieve_offset = PAGE_OFFSET(pgno, pFD->szPage); int64_t retrieve_offset = PAGE_OFFSET(pgno, pFD->szPage);
int64_t pgnoEnd = pgno - 1 + (bOffset + size - n + szPgCont - 1) / szPgCont; int64_t pgnoEnd = pgno - 1 + (bOffset + size - n + szPgCont - 1) / szPgCont;
if (szHint > 0) {
pgnoEnd = pgno - 1 + (bOffset + szHint - n + szPgCont - 1) / szPgCont;
}
int64_t retrieve_size = (pgnoEnd - pgno + 1) * pFD->szPage; int64_t retrieve_size = (pgnoEnd - pgno + 1) * pFD->szPage;
code = s3GetObjectBlock(pFD->objName, retrieve_offset, retrieve_size, 1, &pBlock); code = s3GetObjectBlock(pFD->objName, retrieve_offset, retrieve_size, 1, &pBlock);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -350,6 +355,9 @@ static int32_t tsdbReadFileS3(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64
for (int i = 0; i < nPage; ++i) { for (int i = 0; i < nPage; ++i) {
tsdbCacheSetPageS3(pFD->pTsdb->pgCache, pFD, pgno, pBlock + i * pFD->szPage); tsdbCacheSetPageS3(pFD->pTsdb->pgCache, pFD, pgno, pBlock + i * pFD->szPage);
if (szHint > 0 && n >= size) {
continue;
}
memcpy(pFD->pBuf, pBlock + i * pFD->szPage, pFD->szPage); memcpy(pFD->pBuf, pBlock + i * pFD->szPage, pFD->szPage);
// check // check
@ -375,7 +383,7 @@ _exit:
return code; return code;
} }
int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) { int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size, int64_t szHint) {
int32_t code = 0; int32_t code = 0;
if (!pFD->pFD) { if (!pFD->pFD) {
code = tsdbOpenFileImpl(pFD); code = tsdbOpenFileImpl(pFD);
@ -385,7 +393,7 @@ int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size)
} }
if (pFD->s3File && tsS3BlockSize < 0) { if (pFD->s3File && tsS3BlockSize < 0) {
return tsdbReadFileS3(pFD, offset, pBuf, size); return tsdbReadFileS3(pFD, offset, pBuf, size, szHint);
} else { } else {
return tsdbReadFileImp(pFD, offset, pBuf, size); return tsdbReadFileImp(pFD, offset, pBuf, size);
} }
@ -1141,7 +1149,7 @@ int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) {
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size); code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size, 0);
if (code) goto _err; if (code) goto _err;
// decode // decode
@ -1178,7 +1186,7 @@ int32_t tsdbReadSttBlk(SDataFReader *pReader, int32_t iStt, SArray *aSttBlk) {
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->aSttFD[iStt], offset, pReader->aBuf[0], size); code = tsdbReadFile(pReader->aSttFD[iStt], offset, pReader->aBuf[0], size, 0);
if (code) goto _err; if (code) goto _err;
// decode // decode
@ -1211,7 +1219,7 @@ int32_t tsdbReadDataBlk(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *m
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size); code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size, 0);
if (code) goto _err; if (code) goto _err;
// decode // decode
@ -1242,7 +1250,7 @@ int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aCol
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], pSmaInfo->size); code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], pSmaInfo->size, 0);
if (code) goto _err; if (code) goto _err;
// decode // decode
@ -1276,7 +1284,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo
code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey); code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey);
if (code) goto _err; if (code) goto _err;
code = tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey); code = tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey, 0);
if (code) goto _err; if (code) goto _err;
SDiskDataHdr hdr; SDiskDataHdr hdr;
@ -1322,7 +1330,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo
code = tRealloc(&pReader->aBuf[0], hdr.szBlkCol); code = tRealloc(&pReader->aBuf[0], hdr.szBlkCol);
if (code) goto _err; if (code) goto _err;
code = tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol); code = tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol, 0);
if (code) goto _err; if (code) goto _err;
} }
@ -1366,7 +1374,7 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo
code = tRealloc(&pReader->aBuf[1], size); code = tRealloc(&pReader->aBuf[1], size);
if (code) goto _err; if (code) goto _err;
code = tsdbReadFile(pFD, offset, pReader->aBuf[1], size); code = tsdbReadFile(pFD, offset, pReader->aBuf[1], size, 0);
if (code) goto _err; if (code) goto _err;
code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]); code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]);
@ -1392,7 +1400,7 @@ int32_t tsdbReadDataBlockEx(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockDat
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pDataFD, pBlockInfo->offset, pReader->aBuf[0], pBlockInfo->szBlock); code = tsdbReadFile(pReader->pDataFD, pBlockInfo->offset, pReader->aBuf[0], pBlockInfo->szBlock, 0);
if (code) goto _err; if (code) goto _err;
// decmpr // decmpr
@ -1444,7 +1452,7 @@ int32_t tsdbReadSttBlockEx(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
// read // read
code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock); code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock, 0);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
// decmpr // decmpr
@ -1700,7 +1708,7 @@ int32_t tsdbReadDelDatav1(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelDa
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size); code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size, 0);
if (code) goto _err; if (code) goto _err;
// // decode // // decode
@ -1740,7 +1748,7 @@ int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) {
if (code) goto _err; if (code) goto _err;
// read // read
code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size); code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size, 0);
if (code) goto _err; if (code) goto _err;
// decode // decode