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

Merge pull request #2995 from taosdata/feature/2.0tsdb 

Feature/2.0tsdb
This commit is contained in:
Shengliang Guan 2020-08-10 17:05:51 +08:00 committed by GitHub
parent 5c73b49b83 d7cc7d9ec2
commit 22cbd9627c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 38 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/tsdb/inc/tsdbMain.h
View File

@ -151,22 +151,22 @@ typedef struct {
// ------------------ tsdbFile.c
extern const char* tsdbFileSuffix[];
typedef enum {
#ifdef TSDB_IDX
TSDB_FILE_TYPE_IDX = 0,
TSDB_FILE_TYPE_HEAD,
#else
TSDB_FILE_TYPE_HEAD = 0,
#endif
TSDB_FILE_TYPE_DATA,
TSDB_FILE_TYPE_LAST,
TSDB_FILE_TYPE_MAX,
#ifdef TSDB_IDX
TSDB_FILE_TYPE_NIDX,
#endif
TSDB_FILE_TYPE_STAT,
TSDB_FILE_TYPE_NHEAD,
TSDB_FILE_TYPE_NLAST
TSDB_FILE_TYPE_NDATA,
TSDB_FILE_TYPE_NLAST,
TSDB_FILE_TYPE_NSTAT
} TSDB_FILE_TYPE;
#ifndef TDINTERNAL
#define TSDB_FILE_TYPE_MAX (TSDB_FILE_TYPE_LAST+1)
#else
#define TSDB_FILE_TYPE_MAX (TSDB_FILE_TYPE_STAT+1)
#endif
typedef struct {
uint32_t magic;
uint32_t len;
@ -281,9 +281,6 @@ typedef struct {
TSKEY minKey;
TSKEY maxKey;
SFileGroup fGroup;
#ifdef TSDB_IDX
SFile nIdxF;
#endif
SFile nHeadF;
SFile nLastF;
} SHelperFile;
@ -497,10 +494,6 @@ void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TS
#define helperState(h) (h)->state
#define TSDB_NLAST_FILE_OPENED(h) ((h)->files.nLastF.fd > 0)
#define helperFileId(h) ((h)->files.fGroup.fileId)
#ifdef TSDB_IDX
#define helperIdxF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_IDX]))
#define helperNewIdxF(h) (&((h)->files.nIdxF))
#endif
#define helperHeadF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_HEAD]))
#define helperDataF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_DATA]))
#define helperLastF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_LAST]))
@ -512,7 +505,7 @@ int tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
void tsdbDestroyHelper(SRWHelper* pHelper);
void tsdbResetHelper(SRWHelper* pHelper);
int tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
int tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
int tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError, SFileGroup* pGroup);
int tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
int tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols* pDataCols, TSKEY maxKey);
int tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);

10
src/tsdb/src/tsdbFile.c
View File

@ -21,11 +21,7 @@
#define TAOS_RANDOM_FILE_FAIL_TEST
#ifdef TSDB_IDX
const char *tsdbFileSuffix[] = {".idx", ".head", ".data", ".last", "", ".i", ".h", ".l"};
#else
const char *tsdbFileSuffix[] = {".head", ".data", ".last", "", ".h", ".l"};
#endif
const char *tsdbFileSuffix[] = {".head", ".data", ".last", ".stat", ".h", ".d", ".l", ".s"};
static int tsdbInitFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type);
static void tsdbDestroyFile(SFile *pFile);
@ -413,6 +409,10 @@ static int tsdbInitFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type) {
pBuf = taosDecodeFixedU32(pBuf, &version);
pBuf = tsdbDecodeSFileInfo(pBuf, &(pFile->info));
if (pFile->info.size == TSDB_FILE_HEAD_SIZE) {
pFile->info.size = lseek(pFile->fd, 0, SEEK_END);
}
if (version != TSDB_FILE_VERSION) {
tsdbError("vgId:%d file %s version %u is not the same as program version %u which may cause problem",
REPO_ID(pRepo), pFile->fname, version, TSDB_FILE_VERSION);

9
src/tsdb/src/tsdbMemTable.c
View File

@ -679,15 +679,10 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
}
taosTFree(dataDir);
tsdbCloseHelperFile(pHelper, 0);
tsdbCloseHelperFile(pHelper, 0, pGroup);
pthread_rwlock_wrlock(&(pFileH->fhlock));
#ifdef TSDB_IDX
rename(helperNewIdxF(pHelper)->fname, helperIdxF(pHelper)->fname);
pGroup->files[TSDB_FILE_TYPE_IDX].info = helperNewIdxF(pHelper)->info;
#endif
rename(helperNewHeadF(pHelper)->fname, helperHeadF(pHelper)->fname);
pGroup->files[TSDB_FILE_TYPE_HEAD].info = helperNewHeadF(pHelper)->info;
@ -706,7 +701,7 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
_err:
taosTFree(dataDir);
tsdbCloseHelperFile(pHelper, 1);
tsdbCloseHelperFile(pHelper, 1, NULL);
return -1;
}

76
src/tsdb/src/tsdbRWHelper.c
View File

@ -91,7 +91,7 @@ void tsdbResetHelper(SRWHelper *pHelper) {
tsdbResetHelperTableImpl(pHelper);
// Reset the file part
tsdbCloseHelperFile(pHelper, false);
tsdbCloseHelperFile(pHelper, false, NULL);
tsdbResetHelperFileImpl(pHelper);
pHelper->state = TSDB_HELPER_CLEAR_STATE;
@ -110,31 +110,16 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
// Set the files
pHelper->files.fGroup = *pGroup;
if (helperType(pHelper) == TSDB_WRITE_HELPER) {
#ifdef TSDB_IDX
tsdbGetDataFileName(pHelper->pRepo, pGroup->fileId, TSDB_FILE_TYPE_NIDX, helperNewIdxF(pHelper)->fname);
#endif
tsdbGetDataFileName(pHelper->pRepo, pGroup->fileId, TSDB_FILE_TYPE_NHEAD, helperNewHeadF(pHelper)->fname);
tsdbGetDataFileName(pHelper->pRepo, pGroup->fileId, TSDB_FILE_TYPE_NLAST, helperNewLastF(pHelper)->fname);
}
// Open the files
#ifdef TSDB_IDX
if (tsdbOpenFile(helperIdxF(pHelper), O_RDONLY) < 0) return -1;
#endif
if (tsdbOpenFile(helperHeadF(pHelper), O_RDONLY) < 0) return -1;
if (helperType(pHelper) == TSDB_WRITE_HELPER) {
if (tsdbOpenFile(helperDataF(pHelper), O_RDWR) < 0) return -1;
if (tsdbOpenFile(helperLastF(pHelper), O_RDWR) < 0) return -1;
#ifdef TSDB_IDX
// Create and open .i file
pFile = helperNewIdxF(pHelper);
if (tsdbOpenFile(pFile, O_WRONLY | O_CREAT) < 0) return -1;
pFile->info.size = TSDB_FILE_HEAD_SIZE;
pFile->info.magic = TSDB_FILE_INIT_MAGIC;
if (tsdbUpdateFileHeader(pFile) < 0) return -1;
#endif
// Create and open .h
pFile = helperNewHeadF(pHelper);
if (tsdbOpenFile(pFile, O_WRONLY | O_CREAT) < 0) return -1;
@ -161,14 +146,9 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
return 0;
}
int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError, SFileGroup *pGroup) {
SFile *pFile = NULL;
#ifdef TSDB_IDX
pFile = helperIdxF(pHelper);
tsdbCloseFile(pFile);
#endif
pFile = helperHeadF(pHelper);
tsdbCloseFile(pFile);
@ -177,10 +157,11 @@ int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
if (helperType(pHelper) == TSDB_WRITE_HELPER) {
if (!hasError) {
tsdbUpdateFileHeader(pFile);
fsync(pFile->fd);
} else {
// TODO: shrink back to origin
ASSERT(pGroup != NULL);
taosFtruncate(pFile->fd, pGroup->files[TSDB_FILE_TYPE_DATA].info.size);
}
fsync(pFile->fd);
}
tsdbCloseFile(pFile);
}
@ -190,27 +171,16 @@ int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
if (helperType(pHelper) == TSDB_WRITE_HELPER && !TSDB_NLAST_FILE_OPENED(pHelper)) {
if (!hasError) {
tsdbUpdateFileHeader(pFile);
fsync(pFile->fd);
} else {
// TODO: shrink back to origin
ASSERT(pGroup != NULL);
taosFtruncate(pFile->fd, pGroup->files[TSDB_FILE_TYPE_LAST].info.size);
}
fsync(pFile->fd);
}
tsdbCloseFile(pFile);
}
if (helperType(pHelper) == TSDB_WRITE_HELPER) {
#ifdef TSDB_IDX
pFile = helperNewIdxF(pHelper);
if (pFile->fd > 0) {
if (!hasError) {
tsdbUpdateFileHeader(pFile);
fsync(pFile->fd);
}
tsdbCloseFile(pFile);
if (hasError) (void)remove(pFile->fname);
}
#endif
pFile = helperNewHeadF(pHelper);
if (pFile->fd > 0) {
if (!hasError) {
@ -412,10 +382,6 @@ int tsdbWriteCompInfo(SRWHelper *pHelper) {
return -1;
}
#ifdef TSDB_IDX
pFile = helperNewIdxF(pHelper);
#endif
if (taosTSizeof(pHelper->pWIdx) < pFile->info.len + sizeof(SCompIdx) + 12) {
pHelper->pWIdx = taosTRealloc(pHelper->pWIdx, taosTSizeof(pHelper->pWIdx) == 0 ? 1024 : taosTSizeof(pHelper->pWIdx) * 2);
if (pHelper->pWIdx == NULL) {
@ -426,6 +392,9 @@ int tsdbWriteCompInfo(SRWHelper *pHelper) {
void *pBuf = POINTER_SHIFT(pHelper->pWIdx, pFile->info.len);
pFile->info.len += tsdbEncodeSCompIdx(&pBuf, &(pHelper->curCompIdx));
pFile->info.size += pIdx->len;
// ASSERT(pFile->info.size == lseek(pFile->fd, 0, SEEK_CUR));
}
return 0;
@ -435,11 +404,7 @@ int tsdbWriteCompIdx(SRWHelper *pHelper) {
ASSERT(helperType(pHelper) == TSDB_WRITE_HELPER);
off_t offset = 0;
#ifdef TSDB_IDX
SFile *pFile = helperNewIdxF(pHelper);
#else
SFile *pFile = helperNewHeadF(pHelper);
#endif
pFile->info.len += sizeof(TSCKSUM);
if (taosTSizeof(pHelper->pWIdx) < pFile->info.len) {
@ -460,7 +425,7 @@ int tsdbWriteCompIdx(SRWHelper *pHelper) {
return -1;
}
pFile->info.offset = offset;
ASSERT(offset == pFile->info.size);
if (taosTWrite(pFile->fd, (void *)pHelper->pWIdx, pFile->info.len) < (int)pFile->info.len) {
tsdbError("vgId:%d failed to write %d bytes to file %s since %s", REPO_ID(pHelper->pRepo), pFile->info.len,
@ -469,16 +434,16 @@ int tsdbWriteCompIdx(SRWHelper *pHelper) {
return -1;
}
pFile->info.offset = offset;
pFile->info.size += pFile->info.len;
// ASSERT(pFile->info.size == lseek(pFile->fd, 0, SEEK_CUR));
return 0;
}
int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) {
ASSERT(pHelper->state == TSDB_HELPER_FILE_SET_AND_OPEN);
#ifdef TSDB_IDX
SFile *pFile = helperIdxF(pHelper);
#else
SFile *pFile = helperHeadF(pHelper);
#endif
int fd = pFile->fd;
if (!helperHasState(pHelper, TSDB_HELPER_IDX_LOAD)) {
@ -847,6 +812,9 @@ static int tsdbWriteBlockToFile(SRWHelper *pHelper, SFile *pFile, SDataCols *pDa
(int)(pCompBlock->numOfRows), pCompBlock->len, pCompBlock->numOfCols, pCompBlock->keyFirst,
pCompBlock->keyLast);
pFile->info.size += pCompBlock->len;
// ASSERT(pFile->info.size == lseek(pFile->fd, 0, SEEK_CUR));
return 0;
_err:
@ -1052,10 +1020,6 @@ static void tsdbResetHelperFileImpl(SRWHelper *pHelper) {
helperLastF(pHelper)->fd = -1;
helperNewHeadF(pHelper)->fd = -1;
helperNewLastF(pHelper)->fd = -1;
#ifdef TSDB_IDX
helperIdxF(pHelper)->fd = -1;
helperNewIdxF(pHelper)->fd = -1;
#endif
}
static int tsdbInitHelperFile(SRWHelper *pHelper) {
@ -1064,7 +1028,7 @@ static int tsdbInitHelperFile(SRWHelper *pHelper) {
}
static void tsdbDestroyHelperFile(SRWHelper *pHelper) {
tsdbCloseHelperFile(pHelper, false);
tsdbCloseHelperFile(pHelper, false, NULL);
tsdbResetHelperFileImpl(pHelper);
taosTZfree(pHelper->idxH.pIdxArray);
taosTZfree(pHelper->pWIdx);