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

TD-34

This commit is contained in:
hzcheng 2020-03-24 15:10:41 +08:00
parent eed1d76b3f
commit 305523f47a
2 changed files with 41 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/vnode/tsdb/inc/tsdbFile.h
View File

@ -36,11 +36,6 @@ typedef enum {
extern const char *tsdbFileSuffix[]; extern const char *tsdbFileSuffix[];
typedef struct {
int64_t size;
int64_t tombSize;
} SFileInfo;
typedef struct { typedef struct {
int8_t type; int8_t type;
char fname[128]; char fname[128];

50
src/vnode/tsdb/src/tsdbMain.c
View File

@ -782,6 +782,39 @@ static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int max
return numOfRows; return numOfRows;
} }
static void tsdbDestroyTableIters(SSkipListIterator **iters, int maxTables) {
if (iters == NULL) return;
for (int tid = 0; tid < maxTables; tid++) {
if (iters[tid] == NULL) continue;
tSkipListDestroy(iters[tid]);
}
free(iters);
}
static SSkipListIterator **tsdbCreateTableIters(STsdbMeta *pMeta, int maxTables) {
SSkipListIterator **iters = (SSkipListIterator *)calloc(maxTables, sizeof(SSkipListIterator *));
if (iters == NULL) return NULL;
for (int tid = 0; tid < maxTables; tid++) {
STable *pTable = pMeta->tables[tid];
if (pTable == NULL || pTable->imem == NULL) continue;
iters[tid] = tSkipListCreateIter(pTable->imem->pData);
if (iters[tid] == NULL) {
tsdbDestroyTableIters(iters, maxTables);
return NULL;
}
if (!tSkipListIterNext(iters[tid])) {
assert(false);
}
}
return iters;
}
// Commit to file // Commit to file
static void *tsdbCommitToFile(void *arg) { static void *tsdbCommitToFile(void *arg) {
// TODO // TODO
@ -791,10 +824,8 @@ static void *tsdbCommitToFile(void *arg) {
STsdbCfg * pCfg = &(pRepo->config); STsdbCfg * pCfg = &(pRepo->config);
if (pCache->imem == NULL) return; if (pCache->imem == NULL) return;
int sfid = tsdbGetKeyFileId(pCache->imem->keyFirst, pCfg->daysPerFile, pCfg->precision); // Create the iterator to read from cache
int efid = tsdbGetKeyFileId(pCache->imem->keyLast, pCfg->daysPerFile, pCfg->precision); SSkipListIterator **iters = tsdbCreateTableIters(pMeta, pCfg->maxTables);
SSkipListIterator **iters = (SSkipListIterator **)calloc(pCfg->maxTables, sizeof(SSkipListIterator *));
if (iters == NULL) { if (iters == NULL) {
// TODO: deal with the error // TODO: deal with the error
return NULL; return NULL;
@ -805,10 +836,15 @@ static void *tsdbCommitToFile(void *arg) {
SDataCol **cols = (SDataCol **)malloc(sizeof(SDataCol *) * maxCols); SDataCol **cols = (SDataCol **)malloc(sizeof(SDataCol *) * maxCols);
void *buf = malloc((maxBytes + sizeof(SDataCol)) * pCfg->maxRowsPerFileBlock); void *buf = malloc((maxBytes + sizeof(SDataCol)) * pCfg->maxRowsPerFileBlock);
int sfid = tsdbGetKeyFileId(pCache->imem->keyFirst, pCfg->daysPerFile, pCfg->precision);
int efid = tsdbGetKeyFileId(pCache->imem->keyLast, pCfg->daysPerFile, pCfg->precision);
for (int fid = sfid; fid <= efid; fid++) { for (int fid = sfid; fid <= efid; fid++) {
TSKEY minKey = 0, maxKey = 0; TSKEY minKey = 0, maxKey = 0;
tsdbGetKeyRangeOfFileId(pCfg->daysPerFile, pCfg->precision, fid, &minKey, &maxKey); tsdbGetKeyRangeOfFileId(pCfg->daysPerFile, pCfg->precision, fid, &minKey, &maxKey);
// tsdbOpenFileForWrite(pRepo, fid);
for (int tid = 0; tid < pCfg->maxTables; tid++) { for (int tid = 0; tid < pCfg->maxTables; tid++) {
STable *pTable = pMeta->tables[tid]; STable *pTable = pMeta->tables[tid];
if (pTable == NULL || pTable->imem == NULL) continue; if (pTable == NULL || pTable->imem == NULL) continue;
@ -837,14 +873,10 @@ static void *tsdbCommitToFile(void *arg) {
} }
} }
// Free the iterator tsdbDestroyTableIters(iters, pCfg->maxTables);
for (int tid = 0; tid < pCfg->maxTables; tid++) {
if (iters[tid] != NULL) tSkipListDestroyIter(iters[tid]);
}
free(buf); free(buf);
free(cols); free(cols);
free(iters);
tsdbLockRepo(arg); tsdbLockRepo(arg);
tdListMove(pCache->imem->list, pCache->pool.memPool); tdListMove(pCache->imem->list, pCache->pool.memPool);