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

tsdb/cache: move data file reader for fs iter to cache reader

This commit is contained in:
Minglei Jin 2023-07-20 09:11:26 +08:00
parent 94d8286fad
commit 48222f5dc3
3 changed files with 39 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
source/dnode/vnode/src/inc/tsdb.h
View File

@ -712,7 +712,7 @@ typedef struct SSttBlockLoadInfo {
SBlockData blockData[2]; SBlockData blockData[2];
void *pSttStatisBlkArray; void *pSttStatisBlkArray;
SArray *aSttBlk; SArray *aSttBlk;
int32_t blockIndex[2]; // to denote the loaded block in the corresponding position. int32_t blockIndex[2]; // to denote the loaded block in the corresponding position.
int32_t currentLoadBlockIndex; int32_t currentLoadBlockIndex;
int32_t loadBlocks; int32_t loadBlocks;
double elapsedTime; double elapsedTime;
@ -855,28 +855,31 @@ typedef struct {
SArray *pTombData; SArray *pTombData;
} STableLoadInfo; } STableLoadInfo;
struct SDataFileReader;
typedef struct SCacheRowsReader { typedef struct SCacheRowsReader {
STsdb *pTsdb; STsdb *pTsdb;
STsdbReaderInfo info; STsdbReaderInfo info;
TdThreadMutex readerMutex; TdThreadMutex readerMutex;
SVnode *pVnode; SVnode *pVnode;
STSchema *pSchema; STSchema *pSchema;
STSchema *pCurrSchema; STSchema *pCurrSchema;
uint64_t uid; uint64_t uid;
char **transferBuf; // todo remove it soon char **transferBuf; // todo remove it soon
int32_t numOfCols; int32_t numOfCols;
SArray *pCidList; SArray *pCidList;
int32_t *pSlotIds; int32_t *pSlotIds;
int32_t type; int32_t type;
int32_t tableIndex; // currently returned result tables int32_t tableIndex; // currently returned result tables
STableKeyInfo *pTableList; // table id list STableKeyInfo *pTableList; // table id list
int32_t numOfTables; int32_t numOfTables;
uint64_t *uidList; uint64_t *uidList;
SSHashObj *pTableMap; SSHashObj *pTableMap;
SArray *pLDataIterArray; SArray *pLDataIterArray;
STsdbReadSnap *pReadSnap; struct SDataFileReader *pFileReader;
char *idstr; STsdbReadSnap *pReadSnap;
int64_t lastTs; char *idstr;
int64_t lastTs;
} SCacheRowsReader; } SCacheRowsReader;
typedef struct { typedef struct {

16
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -1900,7 +1900,6 @@ typedef struct SFSNextRowIter {
int32_t iFileSet; int32_t iFileSet;
STFileSet *pFileSet; STFileSet *pFileSet;
TFileSetArray *aDFileSet; TFileSetArray *aDFileSet;
SDataFileReader *pFileReader;
SArray *pIndexList; SArray *pIndexList;
int32_t iBrinIndex; int32_t iBrinIndex;
SBrinBlock brinBlock; SBrinBlock brinBlock;
@ -1970,7 +1969,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
filesName[3] = pFileObj[3]->fname; filesName[3] = pFileObj[3]->fname;
} }
code = tsdbDataFileReaderOpen(filesName, &conf, &state->pFileReader); code = tsdbDataFileReaderOpen(filesName, &conf, &state->pr->pFileReader);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _err; goto _err;
} }
@ -1983,7 +1982,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
state->pLastIter = &state->lastIter; state->pLastIter = &state->lastIter;
loadDataTomb(state->pr, state->pFileReader); loadDataTomb(state->pr, state->pr->pFileReader);
if (!state->pIndexList) { if (!state->pIndexList) {
state->pIndexList = taosArrayInit(1, sizeof(SBrinBlk)); state->pIndexList = taosArrayInit(1, sizeof(SBrinBlk));
@ -1992,7 +1991,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
} }
const TBrinBlkArray *pBlkArray = NULL; const TBrinBlkArray *pBlkArray = NULL;
int32_t code = tsdbDataFileReadBrinBlk(state->pFileReader, &pBlkArray); int32_t code = tsdbDataFileReadBrinBlk(state->pr->pFileReader, &pBlkArray);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _err; goto _err;
} }
@ -2034,7 +2033,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
pBrinBlk = taosArrayGet(state->pIndexList, state->iBrinIndex); pBrinBlk = taosArrayGet(state->pIndexList, state->iBrinIndex);
} }
code = tsdbDataFileReadBrinBlock(state->pFileReader, pBrinBlk, &state->brinBlock); code = tsdbDataFileReadBrinBlock(state->pr->pFileReader, pBrinBlk, &state->brinBlock);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _err; goto _err;
} }
@ -2079,7 +2078,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
--nCols; --nCols;
++aCols; ++aCols;
} }
code = tsdbDataFileReadBlockDataByColumn(state->pFileReader, pRecord, state->pBlockData, state->pTSchema, aCols, code = tsdbDataFileReadBlockDataByColumn(state->pr->pFileReader, pRecord, state->pBlockData, state->pTSchema, aCols,
nCols); nCols);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _err; goto _err;
@ -2315,8 +2314,9 @@ static void clearLastFileSet(SFSNextRowIter *state) {
state->pBlockData = NULL; state->pBlockData = NULL;
} }
if (state->pFileReader) { if (state->pr->pFileReader) {
tsdbDataFileReaderClose(&state->pFileReader); tsdbDataFileReaderClose(&state->pr->pFileReader);
state->pr->pFileReader = NULL;
} }
if (state->pTSRow) { if (state->pTSRow) {

6
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -17,6 +17,7 @@
#include "tarray.h" #include "tarray.h"
#include "tcommon.h" #include "tcommon.h"
#include "tsdb.h" #include "tsdb.h"
#include "tsdbDataFileRW.h"
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t)) #define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
@ -253,6 +254,11 @@ void* tsdbCacherowsReaderClose(void* pReader) {
double elapse = 0; double elapse = 0;
destroySttBlockReader(p->pLDataIterArray, &loadBlocks, &elapse); destroySttBlockReader(p->pLDataIterArray, &loadBlocks, &elapse);
if (p->pFileReader) {
tsdbDataFileReaderClose(&p->pFileReader);
p->pFileReader = NULL;
}
taosMemoryFree((void*)p->idstr); taosMemoryFree((void*)p->idstr);
taosThreadMutexDestroy(&p->readerMutex); taosThreadMutexDestroy(&p->readerMutex);