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

refact code

This commit is contained in:
Hongze Cheng 2022-09-02 10:17:40 +08:00
parent 3286ab2a9e
commit 1132a7f719
9 changed files with 193 additions and 198 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -50,7 +50,7 @@ typedef struct SBlockData SBlockData;
typedef struct SDelFile SDelFile;
typedef struct SHeadFile SHeadFile;
typedef struct SDataFile SDataFile;
typedef struct SLastFile SLastFile;
typedef struct SSstFile SSstFile;
typedef struct SSmaFile SSmaFile;
typedef struct SDFileSet SDFileSet;
typedef struct SDataFWriter SDataFWriter;
@ -219,7 +219,7 @@ bool tsdbDelFileIsSame(SDelFile *pDelFile1, SDelFile *pDelFile2);
int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype);
int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile);
int32_t tPutDataFile(uint8_t *p, SDataFile *pDataFile);
int32_t tPutLastFile(uint8_t *p, SLastFile *pLastFile);
int32_t tPutSstFile(uint8_t *p, SSstFile *pSstFile);
int32_t tPutSmaFile(uint8_t *p, SSmaFile *pSmaFile);
int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile);
int32_t tGetDelFile(uint8_t *p, SDelFile *pDelFile);
@ -228,7 +228,7 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet);
void tsdbHeadFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SHeadFile *pHeadF, char fname[]);
void tsdbDataFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SDataFile *pDataF, char fname[]);
void tsdbLastFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SLastFile *pLastF, char fname[]);
void tsdbSstFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSstFile *pSstF, char fname[]);
void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, char fname[]);
// SDelFile
void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]);
@ -541,7 +541,7 @@ struct SDataFile {
int64_t size;
};
struct SLastFile {
struct SSstFile {
volatile int32_t nRef;
int64_t commitID;
@ -562,8 +562,8 @@ struct SDFileSet {
SHeadFile *pHeadF;
SDataFile *pDataF;
SSmaFile *pSmaF;
uint8_t nLastF;
SLastFile *aLastF[TSDB_MAX_LAST_FILE];
uint8_t nSstF;
SSstFile *aSstF[TSDB_MAX_LAST_FILE];
};
struct SRowIter {
@ -598,7 +598,7 @@ struct SDataFWriter {
SHeadFile fHead;
SDataFile fData;
SSmaFile fSma;
SLastFile fLast[TSDB_MAX_LAST_FILE];
SSstFile fSst[TSDB_MAX_LAST_FILE];
uint8_t *aBuf[4];
};

28
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -431,9 +431,9 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) {
pCommitter->toLastOnly = 0;
SDataFReader *pReader = pCommitter->dReader.pReader;
if (pReader) {
if (pReader->pSet->nLastF >= pCommitter->maxLast) {
if (pReader->pSet->nSstF >= pCommitter->maxLast) {
int8_t iIter = 0;
for (int32_t iLast = 0; iLast < pReader->pSet->nLastF; iLast++) {
for (int32_t iLast = 0; iLast < pReader->pSet->nSstF; iLast++) {
pIter = &pCommitter->aDataIter[iIter];
pIter->type = LAST_DATA_ITER;
pIter->iLast = iLast;
@ -457,9 +457,9 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) {
iIter++;
}
} else {
for (int32_t iLast = 0; iLast < pReader->pSet->nLastF; iLast++) {
SLastFile *pLastFile = pReader->pSet->aLastF[iLast];
if (pLastFile->size > pLastFile->offset) {
for (int32_t iLast = 0; iLast < pReader->pSet->nSstF; iLast++) {
SSstFile *pSstFile = pReader->pSet->aSstF[iLast];
if (pSstFile->size > pSstFile->offset) {
pCommitter->toLastOnly = 1;
break;
}
@ -515,29 +515,29 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
SHeadFile fHead = {.commitID = pCommitter->commitID};
SDataFile fData = {.commitID = pCommitter->commitID};
SSmaFile fSma = {.commitID = pCommitter->commitID};
SLastFile fLast = {.commitID = pCommitter->commitID};
SSstFile fSst = {.commitID = pCommitter->commitID};
SDFileSet wSet = {.fid = pCommitter->commitFid, .pHeadF = &fHead, .pDataF = &fData, .pSmaF = &fSma};
if (pRSet) {
ASSERT(pRSet->nLastF <= pCommitter->maxLast);
ASSERT(pRSet->nSstF <= pCommitter->maxLast);
fData = *pRSet->pDataF;
fSma = *pRSet->pSmaF;
wSet.diskId = pRSet->diskId;
if (pRSet->nLastF < pCommitter->maxLast) {
for (int32_t iLast = 0; iLast < pRSet->nLastF; iLast++) {
wSet.aLastF[iLast] = pRSet->aLastF[iLast];
if (pRSet->nSstF < pCommitter->maxLast) {
for (int32_t iLast = 0; iLast < pRSet->nSstF; iLast++) {
wSet.aSstF[iLast] = pRSet->aSstF[iLast];
}
wSet.nLastF = pRSet->nLastF + 1;
wSet.nSstF = pRSet->nSstF + 1;
} else {
wSet.nLastF = 1;
wSet.nSstF = 1;
}
} else {
SDiskID did = {0};
tfsAllocDisk(pTsdb->pVnode->pTfs, 0, &did);
tfsMkdirRecurAt(pTsdb->pVnode->pTfs, pTsdb->path, did);
wSet.diskId = did;
wSet.nLastF = 1;
wSet.nSstF = 1;
}
wSet.aLastF[wSet.nLastF - 1] = &fLast;
wSet.aSstF[wSet.nSstF - 1] = &fSst;
code = tsdbDataFWriterOpen(&pCommitter->dWriter.pWriter, pTsdb, &wSet);
if (code) goto _err;

168
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -255,8 +255,8 @@ void tsdbFSDestroy(STsdbFS *pFS) {
taosMemoryFree(pSet->pHeadF);
taosMemoryFree(pSet->pDataF);
taosMemoryFree(pSet->pSmaF);
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
taosMemoryFree(pSet->aLastF[iLast]);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
taosMemoryFree(pSet->aSstF[iLast]);
}
}
@ -312,12 +312,12 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
}
// last ===========
tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aLastF[0], fname);
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[0], fname);
if (taosStatFile(fname, &size, NULL)) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (size != pSet->aLastF[0]->size) {
if (size != pSet->aSstF[0]->size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
@ -509,8 +509,8 @@ int32_t tsdbFSClose(STsdb *pTsdb) {
taosMemoryFree(pSet->pDataF);
// last
ASSERT(pSet->aLastF[0]->nRef == 1);
taosMemoryFree(pSet->aLastF[0]);
ASSERT(pSet->aSstF[0]->nRef == 1);
taosMemoryFree(pSet->aSstF[0]);
// sma
ASSERT(pSet->pSmaF->nRef == 1);
@ -571,13 +571,13 @@ int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS) {
*fSet.pSmaF = *pSet->pSmaF;
// last
for (fSet.nLastF = 0; fSet.nLastF < pSet->nLastF; fSet.nLastF++) {
fSet.aLastF[fSet.nLastF] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (fSet.aLastF[fSet.nLastF] == NULL) {
for (fSet.nSstF = 0; fSet.nSstF < pSet->nSstF; fSet.nSstF++) {
fSet.aSstF[fSet.nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (fSet.aSstF[fSet.nSstF] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.aLastF[fSet.nLastF] = *pSet->aLastF[fSet.nLastF];
*fSet.aSstF[fSet.nSstF] = *pSet->aSstF[fSet.nSstF];
}
if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
@ -631,27 +631,27 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
*pDFileSet->pDataF = *pSet->pDataF;
*pDFileSet->pSmaF = *pSet->pSmaF;
// last
if (pSet->nLastF > pDFileSet->nLastF) {
ASSERT(pSet->nLastF == pDFileSet->nLastF + 1);
if (pSet->nSstF > pDFileSet->nSstF) {
ASSERT(pSet->nSstF == pDFileSet->nSstF + 1);
pDFileSet->aLastF[pDFileSet->nLastF] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (pDFileSet->aLastF[pDFileSet->nLastF] == NULL) {
pDFileSet->aSstF[pDFileSet->nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (pDFileSet->aSstF[pDFileSet->nSstF] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*pDFileSet->aLastF[pDFileSet->nLastF] = *pSet->aLastF[pSet->nLastF - 1];
pDFileSet->nLastF++;
} else if (pSet->nLastF < pDFileSet->nLastF) {
ASSERT(pSet->nLastF == 1);
for (int32_t iLast = 1; iLast < pDFileSet->nLastF; iLast++) {
taosMemoryFree(pDFileSet->aLastF[iLast]);
*pDFileSet->aSstF[pDFileSet->nSstF] = *pSet->aSstF[pSet->nSstF - 1];
pDFileSet->nSstF++;
} else if (pSet->nSstF < pDFileSet->nSstF) {
ASSERT(pSet->nSstF == 1);
for (int32_t iLast = 1; iLast < pDFileSet->nSstF; iLast++) {
taosMemoryFree(pDFileSet->aSstF[iLast]);
}
*pDFileSet->aLastF[0] = *pSet->aLastF[0];
pDFileSet->nLastF = 1;
*pDFileSet->aSstF[0] = *pSet->aSstF[0];
pDFileSet->nSstF = 1;
} else {
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
*pDFileSet->aLastF[iLast] = *pSet->aLastF[iLast];
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
*pDFileSet->aSstF[iLast] = *pSet->aSstF[iLast];
}
}
@ -659,8 +659,8 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
}
}
ASSERT(pSet->nLastF == 1);
SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid, .nLastF = 1};
ASSERT(pSet->nSstF == 1);
SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid, .nSstF = 1};
// head
fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
@ -687,12 +687,12 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
*fSet.pSmaF = *pSet->pSmaF;
// last
fSet.aLastF[0] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (fSet.aLastF[0] == NULL) {
fSet.aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (fSet.aSstF[0] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.aLastF[0] = *pSet->aLastF[0];
*fSet.aSstF[0] = *pSet->aSstF[0];
if (taosArrayInsert(pFS->aDFileSet, idx, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
@ -862,74 +862,74 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
// last
if (sameDisk) {
if (pSetNew->nLastF > pSetOld->nLastF) {
ASSERT(pSetNew->nLastF = pSetOld->nLastF + 1);
pSetOld->aLastF[pSetOld->nLastF] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (pSetOld->aLastF[pSetOld->nLastF] == NULL) {
if (pSetNew->nSstF > pSetOld->nSstF) {
ASSERT(pSetNew->nSstF = pSetOld->nSstF + 1);
pSetOld->aSstF[pSetOld->nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (pSetOld->aSstF[pSetOld->nSstF] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*pSetOld->aLastF[pSetOld->nLastF] = *pSetNew->aLastF[pSetOld->nLastF];
pSetOld->aLastF[pSetOld->nLastF]->nRef = 1;
pSetOld->nLastF++;
} else if (pSetNew->nLastF < pSetOld->nLastF) {
ASSERT(pSetNew->nLastF == 1);
for (int32_t iLast = 0; iLast < pSetOld->nLastF; iLast++) {
SLastFile *pLastFile = pSetOld->aLastF[iLast];
nRef = atomic_sub_fetch_32(&pLastFile->nRef, 1);
*pSetOld->aSstF[pSetOld->nSstF] = *pSetNew->aSstF[pSetOld->nSstF];
pSetOld->aSstF[pSetOld->nSstF]->nRef = 1;
pSetOld->nSstF++;
} else if (pSetNew->nSstF < pSetOld->nSstF) {
ASSERT(pSetNew->nSstF == 1);
for (int32_t iLast = 0; iLast < pSetOld->nSstF; iLast++) {
SSstFile *pSstFile = pSetOld->aSstF[iLast];
nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
if (nRef == 0) {
tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pLastFile, fname);
tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
taosRemoveFile(fname);
taosMemoryFree(pLastFile);
taosMemoryFree(pSstFile);
}
pSetOld->aLastF[iLast] = NULL;
pSetOld->aSstF[iLast] = NULL;
}
pSetOld->nLastF = 1;
pSetOld->aLastF[0] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (pSetOld->aLastF[0] == NULL) {
pSetOld->nSstF = 1;
pSetOld->aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (pSetOld->aSstF[0] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*pSetOld->aLastF[0] = *pSetNew->aLastF[0];
pSetOld->aLastF[0]->nRef = 1;
*pSetOld->aSstF[0] = *pSetNew->aSstF[0];
pSetOld->aSstF[0]->nRef = 1;
} else {
for (int32_t iLast = 0; iLast < pSetOld->nLastF; iLast++) {
SLastFile *pLastFile = pSetOld->aLastF[iLast];
nRef = atomic_sub_fetch_32(&pLastFile->nRef, 1);
for (int32_t iLast = 0; iLast < pSetOld->nSstF; iLast++) {
SSstFile *pSstFile = pSetOld->aSstF[iLast];
nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
if (nRef == 0) {
tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pLastFile, fname);
tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
taosRemoveFile(fname);
taosMemoryFree(pLastFile);
taosMemoryFree(pSstFile);
}
pSetOld->aLastF[iLast] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (pSetOld->aLastF[iLast] == NULL) {
pSetOld->aSstF[iLast] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (pSetOld->aSstF[iLast] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*pSetOld->aLastF[iLast] = *pSetNew->aLastF[iLast];
pSetOld->aLastF[iLast]->nRef = 1;
*pSetOld->aSstF[iLast] = *pSetNew->aSstF[iLast];
pSetOld->aSstF[iLast]->nRef = 1;
}
}
} else {
ASSERT(pSetOld->nLastF == pSetNew->nLastF);
for (int32_t iLast = 0; iLast < pSetOld->nLastF; iLast++) {
SLastFile *pLastFile = pSetOld->aLastF[iLast];
nRef = atomic_sub_fetch_32(&pLastFile->nRef, 1);
ASSERT(pSetOld->nSstF == pSetNew->nSstF);
for (int32_t iLast = 0; iLast < pSetOld->nSstF; iLast++) {
SSstFile *pSstFile = pSetOld->aSstF[iLast];
nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
if (nRef == 0) {
tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pLastFile, fname);
tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
taosRemoveFile(fname);
taosMemoryFree(pLastFile);
taosMemoryFree(pSstFile);
}
pSetOld->aLastF[iLast] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (pSetOld->aLastF[iLast] == NULL) {
pSetOld->aSstF[iLast] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (pSetOld->aSstF[iLast] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*pSetOld->aLastF[iLast] = *pSetNew->aLastF[iLast];
pSetOld->aLastF[iLast]->nRef = 1;
*pSetOld->aSstF[iLast] = *pSetNew->aSstF[iLast];
pSetOld->aSstF[iLast]->nRef = 1;
}
}
@ -963,12 +963,12 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
taosMemoryFree(pSetOld->pSmaF);
}
for (int8_t iLast = 0; iLast < pSetOld->nLastF; iLast++) {
nRef = atomic_sub_fetch_32(&pSetOld->aLastF[iLast]->nRef, 1);
for (int8_t iLast = 0; iLast < pSetOld->nSstF; iLast++) {
nRef = atomic_sub_fetch_32(&pSetOld->aSstF[iLast]->nRef, 1);
if (nRef == 0) {
tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->aLastF[iLast], fname);
tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->aSstF[iLast], fname);
taosRemoveFile(fname);
taosMemoryFree(pSetOld->aLastF[iLast]);
taosMemoryFree(pSetOld->aSstF[iLast]);
}
}
@ -976,7 +976,7 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
continue;
_add_new:
fSet = (SDFileSet){.diskId = pSetNew->diskId, .fid = pSetNew->fid, .nLastF = 1};
fSet = (SDFileSet){.diskId = pSetNew->diskId, .fid = pSetNew->fid, .nSstF = 1};
// head
fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
@ -1006,14 +1006,14 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
fSet.pSmaF->nRef = 1;
// last
ASSERT(pSetNew->nLastF == 1);
fSet.aLastF[0] = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
if (fSet.aLastF[0] == NULL) {
ASSERT(pSetNew->nSstF == 1);
fSet.aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
if (fSet.aSstF[0] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
*fSet.aLastF[0] = *pSetNew->aLastF[0];
fSet.aLastF[0]->nRef = 1;
*fSet.aSstF[0] = *pSetNew->aSstF[0];
fSet.aSstF[0]->nRef = 1;
if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
@ -1061,8 +1061,8 @@ int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS) {
nRef = atomic_fetch_add_32(&pSet->pSmaF->nRef, 1);
ASSERT(nRef > 0);
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
nRef = atomic_fetch_add_32(&pSet->aLastF[iLast]->nRef, 1);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
nRef = atomic_fetch_add_32(&pSet->aSstF[iLast]->nRef, 1);
ASSERT(nRef > 0);
}
@ -1121,13 +1121,13 @@ void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS) {
}
// last
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
nRef = atomic_sub_fetch_32(&pSet->aLastF[iLast]->nRef, 1);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
nRef = atomic_sub_fetch_32(&pSet->aSstF[iLast]->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aLastF[iLast], fname);
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iLast], fname);
taosRemoveFile(fname);
taosMemoryFree(pSet->aLastF[iLast]);
taosMemoryFree(pSet->aSstF[iLast]);
/* code */
}
}

38
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -53,22 +53,22 @@ static int32_t tGetDataFile(uint8_t *p, SDataFile *pDataFile) {
return n;
}
int32_t tPutLastFile(uint8_t *p, SLastFile *pLastFile) {
int32_t tPutSstFile(uint8_t *p, SSstFile *pSstFile) {
int32_t n = 0;
n += tPutI64v(p ? p + n : p, pLastFile->commitID);
n += tPutI64v(p ? p + n : p, pLastFile->size);
n += tPutI64v(p ? p + n : p, pLastFile->offset);
n += tPutI64v(p ? p + n : p, pSstFile->commitID);
n += tPutI64v(p ? p + n : p, pSstFile->size);
n += tPutI64v(p ? p + n : p, pSstFile->offset);
return n;
}
static int32_t tGetLastFile(uint8_t *p, SLastFile *pLastFile) {
static int32_t tGetSstFile(uint8_t *p, SSstFile *pSstFile) {
int32_t n = 0;
n += tGetI64v(p + n, &pLastFile->commitID);
n += tGetI64v(p + n, &pLastFile->size);
n += tGetI64v(p + n, &pLastFile->offset);
n += tGetI64v(p + n, &pSstFile->commitID);
n += tGetI64v(p + n, &pSstFile->size);
n += tGetI64v(p + n, &pSstFile->offset);
return n;
}
@ -102,9 +102,9 @@ void tsdbDataFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SDataFile *pDataF,
TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pDataF->commitID, ".data");
}
void tsdbLastFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SLastFile *pLastF, char fname[]) {
void tsdbSstFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSstFile *pSstF, char fname[]) {
snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s%sv%df%dver%" PRId64 "%s", tfsGetDiskPath(pTsdb->pVnode->pTfs, did),
TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pLastF->commitID, ".last");
TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pSstF->commitID, ".sst");
}
void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, char fname[]) {
@ -195,9 +195,9 @@ int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) {
n += tPutSmaFile(p ? p + n : p, pSet->pSmaF);
// last
n += tPutU8(p ? p + n : p, pSet->nLastF);
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
n += tPutLastFile(p ? p + n : p, pSet->aLastF[iLast]);
n += tPutU8(p ? p + n : p, pSet->nSstF);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
n += tPutSstFile(p ? p + n : p, pSet->aSstF[iLast]);
}
return n;
@ -235,14 +235,14 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
n += tGetSmaFile(p + n, pSet->pSmaF);
// last
n += tGetU8(p + n, &pSet->nLastF);
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
pSet->aLastF[iLast] = (SLastFile *)taosMemoryCalloc(1, sizeof(SLastFile));
if (pSet->aLastF[iLast] == NULL) {
n += tGetU8(p + n, &pSet->nSstF);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
pSet->aSstF[iLast] = (SSstFile *)taosMemoryCalloc(1, sizeof(SSstFile));
if (pSet->aSstF[iLast] == NULL) {
return -1;
}
pSet->aLastF[iLast]->nRef = 1;
n += tGetLastFile(p + n, pSet->aLastF[iLast]);
pSet->aSstF[iLast]->nRef = 1;
n += tGetSstFile(p + n, pSet->aSstF[iLast]);
}
return n;

13
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -244,9 +244,7 @@ _exit:
return (code == TSDB_CODE_SUCCESS) && (pIter->pBlockL != NULL);
}
SRowInfo *tLDataIterGet(SLDataIter *pIter) {
return &pIter->rInfo;
}
SRowInfo *tLDataIterGet(SLDataIter *pIter) { return &pIter->rInfo; }
// SMergeTree =================================================
static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) {
@ -271,7 +269,8 @@ static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) {
}
}
void tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange) {
void tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t uid, STimeWindow *pTimeWindow,
SVersionRange *pVerRange) {
pMTree->backward = backward;
pMTree->pIter = NULL;
pMTree->pIterList = taosArrayInit(4, POINTER_BYTES);
@ -279,7 +278,7 @@ void tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader,
tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);
struct SLDataIter *pIterList[TSDB_DEFAULT_LAST_FILE] = {0};
for(int32_t i = 0; i < pFReader->pSet->nLastF; ++i) { // open all last file
for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file
/*int32_t code = */ tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange);
bool hasVal = tLDataIterNextRow(pIterList[i]);
if (hasVal) {
@ -326,9 +325,7 @@ bool tMergeTreeNext(SMergeTree* pMTree) {
return pMTree->pIter != NULL;
}
TSDBROW tMergeTreeGetRow(SMergeTree* pMTree) {
return pMTree->pIter->rInfo.row;
}
TSDBROW tMergeTreeGetRow(SMergeTree *pMTree) { return pMTree->pIter->rInfo.row; }
void tMergeTreeClose(SMergeTree *pMTree) {
size_t size = taosArrayGetSize(pMTree->pIterList);

20
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -176,10 +176,10 @@ static int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pR
static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order);
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow,
STsdbReader* pReader, bool* freeTSRow);
static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader,
STSRow** pTSRow);
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList,
STSRow** pTSRow, STsdbReader* pReader, bool* freeTSRow);
static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo* pBlockScanInfo,
STsdbReader* pReader, STSRow** pTSRow);
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
STsdbReader* pReader);
@ -617,7 +617,7 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
}
}
pBlockNum->numOfLastFiles = pReader->pFileReader->pSet->nLastF;
pBlockNum->numOfLastFiles = pReader->pFileReader->pSet->nSstF;
int32_t total = pBlockNum->numOfLastFiles + pBlockNum->numOfBlocks;
double el = (taosGetTimestampUs() - st) / 1000.0;
@ -1860,9 +1860,7 @@ static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
return key.ts;
}
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) {
return pLastBlockReader->mergeTree.pIter != NULL;
}
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; }
int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
STsdbReader* pReader) {
@ -1958,7 +1956,6 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
}
bool hasBlockLData = hasDataInLastBlock(pLastBlockReader);
// no data in last block and block, no need to proceed.
@ -2185,7 +2182,7 @@ static int32_t moveToNextFile(STsdbReader* pReader, SBlockNumber* pBlockNum) {
return code;
}
if (taosArrayGetSize(pIndexList) > 0 || pReader->pFileReader->pSet->nLastF > 0) {
if (taosArrayGetSize(pIndexList) > 0 || pReader->pFileReader->pSet->nSstF > 0) {
code = doLoadFileBlock(pReader, pIndexList, pBlockNum);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(pIndexList);
@ -3020,7 +3017,8 @@ int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pR
}
if (pBlockScanInfo->iter.hasVal && pRow != NULL) {
return doMergeMemTableMultiRows(pRow, pBlockScanInfo->uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader, freeTSRow);
return doMergeMemTableMultiRows(pRow, pBlockScanInfo->uid, &pBlockScanInfo->iter, pDelList, pTSRow, pReader,
freeTSRow);
}
if (pBlockScanInfo->iiter.hasVal && piRow != NULL) {

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

@ -434,8 +434,8 @@ int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pS
}
// last
for (int32_t iLast = 0; iLast < pSet->nLastF; iLast++) {
tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aLastF[iLast], fname);
for (int32_t iLast = 0; iLast < pSet->nSstF; iLast++) {
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iLast], fname);
pReader->aLastFD[iLast] = taosOpenFile(fname, TD_FILE_READ);
if (pReader->aLastFD[iLast] == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
@ -475,7 +475,7 @@ int32_t tsdbDataFReaderClose(SDataFReader **ppReader) {
}
// last
for (int32_t iLast = 0; iLast < (*ppReader)->pSet->nLastF; iLast++) {
for (int32_t iLast = 0; iLast < (*ppReader)->pSet->nSstF; iLast++) {
if (taosCloseFile(&(*ppReader)->aLastFD[iLast]) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
@ -561,8 +561,8 @@ _err:
int32_t tsdbReadBlockL(SDataFReader *pReader, int32_t iLast, SArray *aBlockL) {
int32_t code = 0;
int64_t offset = pReader->pSet->aLastF[iLast]->offset;
int64_t size = pReader->pSet->aLastF[iLast]->size - offset;
int64_t offset = pReader->pSet->aSstF[iLast]->offset;
int64_t size = pReader->pSet->aSstF[iLast]->size - offset;
int64_t n;
uint32_t delimiter;
@ -947,14 +947,14 @@ int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pS
.pHeadF = &pWriter->fHead,
.pDataF = &pWriter->fData,
.pSmaF = &pWriter->fSma,
.nLastF = pSet->nLastF //
.nSstF = pSet->nSstF //
};
pWriter->fHead = *pSet->pHeadF;
pWriter->fData = *pSet->pDataF;
pWriter->fSma = *pSet->pSmaF;
for (int8_t iLast = 0; iLast < pSet->nLastF; iLast++) {
pWriter->wSet.aLastF[iLast] = &pWriter->fLast[iLast];
pWriter->fLast[iLast] = *pSet->aLastF[iLast];
for (int8_t iLast = 0; iLast < pSet->nSstF; iLast++) {
pWriter->wSet.aSstF[iLast] = &pWriter->fSst[iLast];
pWriter->fSst[iLast] = *pSet->aSstF[iLast];
}
// head
@ -1037,9 +1037,9 @@ int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pS
}
// last
ASSERT(pWriter->fLast[pSet->nLastF - 1].size == 0);
ASSERT(pWriter->fSst[pSet->nSstF - 1].size == 0);
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
tsdbLastFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fLast[pSet->nLastF - 1], fname);
tsdbSstFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSst[pSet->nSstF - 1], fname);
pWriter->pLastFD = taosOpenFile(fname, flag);
if (pWriter->pLastFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
@ -1050,7 +1050,7 @@ int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pS
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pWriter->fLast[pWriter->wSet.nLastF - 1].size += TSDB_FHDR_SIZE;
pWriter->fSst[pWriter->wSet.nSstF - 1].size += TSDB_FHDR_SIZE;
*ppWriter = pWriter;
return code;
@ -1181,7 +1181,7 @@ int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
// last ==============
memset(hdr, 0, TSDB_FHDR_SIZE);
tPutLastFile(hdr, &pWriter->fLast[pWriter->wSet.nLastF - 1]);
tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_SET);
@ -1303,13 +1303,13 @@ _err:
int32_t tsdbWriteBlockL(SDataFWriter *pWriter, SArray *aBlockL) {
int32_t code = 0;
SLastFile *pLastFile = &pWriter->fLast[pWriter->wSet.nLastF - 1];
SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1];
int64_t size;
int64_t n;
// check
if (taosArrayGetSize(aBlockL) == 0) {
pLastFile->offset = pLastFile->size;
pSstFile->offset = pSstFile->size;
goto _exit;
}
@ -1342,12 +1342,12 @@ int32_t tsdbWriteBlockL(SDataFWriter *pWriter, SArray *aBlockL) {
}
// update
pLastFile->offset = pLastFile->size;
pLastFile->size += size;
pSstFile->offset = pSstFile->size;
pSstFile->size += size;
_exit:
tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
pLastFile->offset, size);
pSstFile->offset, size);
return code;
_err:
@ -1431,7 +1431,7 @@ int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlock
ASSERT(pBlockData->nRow > 0);
pBlkInfo->offset = toLast ? pWriter->fLast[pWriter->wSet.nLastF - 1].size : pWriter->fData.size;
pBlkInfo->offset = toLast ? pWriter->fSst[pWriter->wSet.nSstF - 1].size : pWriter->fData.size;
pBlkInfo->szBlock = 0;
pBlkInfo->szKey = 0;
@ -1475,7 +1475,7 @@ int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlock
// update info
if (toLast) {
pWriter->fLast[pWriter->wSet.nLastF - 1].size += pBlkInfo->szBlock;
pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock;
} else {
pWriter->fData.size += pBlkInfo->szBlock;
}
@ -1555,8 +1555,8 @@ int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
taosCloseFile(&PInFD);
// last
tsdbLastFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->aLastF[0], fNameFrom);
tsdbLastFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->aLastF[0], fNameTo);
tsdbSstFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->aSstF[0], fNameFrom);
tsdbSstFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->aSstF[0], fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
@ -1570,7 +1570,7 @@ int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aLastF[0]->size);
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;

2
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -60,7 +60,7 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
if (expLevel < 0) {
taosMemoryFree(pSet->pHeadF);
taosMemoryFree(pSet->pDataF);
taosMemoryFree(pSet->aLastF[0]);
taosMemoryFree(pSet->aSstF[0]);
taosMemoryFree(pSet->pSmaF);
taosArrayRemove(fs.aDFileSet, iSet);
iSet--;

12
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -931,25 +931,25 @@ static int32_t tsdbSnapWriteData(STsdbSnapWriter* pWriter, uint8_t* pData, uint3
// write
SHeadFile fHead;
SDataFile fData;
SLastFile fLast;
SSstFile fLast;
SSmaFile fSma;
SDFileSet wSet = {.pHeadF = &fHead, .pDataF = &fData, .aLastF[0] = &fLast, .pSmaF = &fSma};
SDFileSet wSet = {.pHeadF = &fHead, .pDataF = &fData, .aSstF[0] = &fLast, .pSmaF = &fSma};
if (pSet) {
wSet.diskId = pSet->diskId;
wSet.fid = fid;
wSet.nLastF = 1;
wSet.nSstF = 1;
fHead = (SHeadFile){.commitID = pWriter->commitID, .offset = 0, .size = 0};
fData = *pSet->pDataF;
fLast = (SLastFile){.commitID = pWriter->commitID, .size = 0};
fLast = (SSstFile){.commitID = pWriter->commitID, .size = 0};
fSma = *pSet->pSmaF;
} else {
wSet.diskId = (SDiskID){.level = 0, .id = 0};
wSet.fid = fid;
wSet.nLastF = 1;
wSet.nSstF = 1;
fHead = (SHeadFile){.commitID = pWriter->commitID, .offset = 0, .size = 0};
fData = (SDataFile){.commitID = pWriter->commitID, .size = 0};
fLast = (SLastFile){.commitID = pWriter->commitID, .size = 0, .offset = 0};
fLast = (SSstFile){.commitID = pWriter->commitID, .size = 0, .offset = 0};
fSma = (SSmaFile){.commitID = pWriter->commitID, .size = 0};
}