Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 0c27b07c 编写于 作者: C Cary Xu
浏览文件
操作

Merge branch '3.0' into feature/3.0_interval_hash_optimize

上级 31fb2a31 07c19b73
隐藏空白更改
内联 并排
Showing with 996 addition and 1233 deletion
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 0c27b07c
...@@ -65,12 +65,14 @@ typedef struct SBlockInfo SBlockInfo; ...@@ -65,12 +65,14 @@ typedef struct SBlockInfo SBlockInfo;
typedef struct SSmaInfo SSmaInfo; typedef struct SSmaInfo SSmaInfo;
typedef struct SBlockCol SBlockCol; typedef struct SBlockCol SBlockCol;
typedef struct SVersionRange SVersionRange; typedef struct SVersionRange SVersionRange;
typedef struct SLDataIter SLDataIter;
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F) #define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_MAX_SUBBLOCKS 8 #define TSDB_MAX_SUBBLOCKS 8
#define TSDB_MAX_LAST_FILE 16 #define TSDB_MAX_SST_FILE 16
#define TSDB_DEFAULT_LAST_FILE 8 #define TSDB_DEFAULT_SST_FILE 8
#define TSDB_FHDR_SIZE 512 #define TSDB_FHDR_SIZE 512
#define TSDB_DEFAULT_PAGE_SIZE 4096
#define HAS_NONE ((int8_t)0x1) #define HAS_NONE ((int8_t)0x1)
#define HAS_NULL ((int8_t)0x2) #define HAS_NULL ((int8_t)0x2)
...@@ -82,6 +84,14 @@ typedef struct SVersionRange SVersionRange; ...@@ -82,6 +84,14 @@ typedef struct SVersionRange SVersionRange;
#define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN}) #define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN})
#define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX}) #define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX})
#define PAGE_CONTENT_SIZE(PAGE) ((PAGE) - sizeof(TSCKSUM))
#define LOGIC_TO_FILE_OFFSET(LOFFSET, PAGE) \
((LOFFSET) / PAGE_CONTENT_SIZE(PAGE) * (PAGE) + (LOFFSET) % PAGE_CONTENT_SIZE(PAGE))
#define FILE_TO_LOGIC_OFFSET(OFFSET, PAGE) ((OFFSET) / (PAGE)*PAGE_CONTENT_SIZE(PAGE) + (OFFSET) % (PAGE))
#define PAGE_OFFSET(PGNO, PAGE) (((PGNO)-1) * (PAGE))
#define OFFSET_PGNO(OFFSET, PAGE) ((OFFSET) / (PAGE) + 1)
#define LOGIC_TO_FILE_SIZE(LSIZE, PAGE) OFFSET_PGNO(LOGIC_TO_FILE_OFFSET(LSIZE, PAGE), PAGE) * (PAGE)
// tsdbUtil.c ============================================================================================== // tsdbUtil.c ==============================================================================================
// TSDBROW // TSDBROW
#define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow]) #define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow])
...@@ -195,7 +205,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol ...@@ -195,7 +205,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
uint8_t **ppBuf); uint8_t **ppBuf);
int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, int32_t nVal, SColData *pColData, int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, int32_t nVal, SColData *pColData,
uint8_t **ppBuf); uint8_t **ppBuf);
int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck);
// tsdbMemTable ============================================================================================== // tsdbMemTable ==============================================================================================
// SMemTable // SMemTable
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable); int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
...@@ -563,7 +572,7 @@ struct SDFileSet { ...@@ -563,7 +572,7 @@ struct SDFileSet {
SDataFile *pDataF; SDataFile *pDataF;
SSmaFile *pSmaF; SSmaFile *pSmaF;
uint8_t nSstF; uint8_t nSstF;
SSstFile *aSstF[TSDB_MAX_LAST_FILE]; SSstFile *aSstF[TSDB_MAX_SST_FILE];
}; };
struct SRowIter { struct SRowIter {
...@@ -578,46 +587,54 @@ struct SRowMerger { ...@@ -578,46 +587,54 @@ struct SRowMerger {
SArray *pArray; // SArray<SColVal> SArray *pArray; // SArray<SColVal>
}; };
struct SDelFWriter { typedef struct {
STsdb *pTsdb; char *path;
SDelFile fDel; int32_t szPage;
TdFilePtr pWriteH; int32_t flag;
TdFilePtr pFD;
int64_t pgno;
uint8_t *pBuf;
int64_t szFile;
} STsdbFD;
struct SDelFWriter {
STsdb *pTsdb;
SDelFile fDel;
STsdbFD *pWriteH;
uint8_t *aBuf[1]; uint8_t *aBuf[1];
}; };
struct STsdbReadSnap {
SMemTable *pMem;
SMemTable *pIMem;
STsdbFS fs;
};
struct SDataFWriter { struct SDataFWriter {
STsdb *pTsdb; STsdb *pTsdb;
SDFileSet wSet; SDFileSet wSet;
TdFilePtr pHeadFD; STsdbFD *pHeadFD;
TdFilePtr pDataFD; STsdbFD *pDataFD;
TdFilePtr pSmaFD; STsdbFD *pSmaFD;
TdFilePtr pLastFD; STsdbFD *pSstFD;
SHeadFile fHead; SHeadFile fHead;
SDataFile fData; SDataFile fData;
SSmaFile fSma; SSmaFile fSma;
SSstFile fSst[TSDB_MAX_LAST_FILE]; SSstFile fSst[TSDB_MAX_SST_FILE];
uint8_t *aBuf[4]; uint8_t *aBuf[4];
}; };
struct STsdbReadSnap {
SMemTable *pMem;
SMemTable *pIMem;
STsdbFS fs;
};
struct SDataFReader { struct SDataFReader {
STsdb *pTsdb; STsdb *pTsdb;
SDFileSet *pSet; SDFileSet *pSet;
TdFilePtr pHeadFD; STsdbFD *pHeadFD;
TdFilePtr pDataFD; STsdbFD *pDataFD;
TdFilePtr pSmaFD; STsdbFD *pSmaFD;
TdFilePtr aLastFD[TSDB_MAX_LAST_FILE]; STsdbFD *aSstFD[TSDB_MAX_SST_FILE];
uint8_t *aBuf[3];
uint8_t *aBuf[3];
}; };
typedef struct { typedef struct {
...@@ -627,15 +644,16 @@ typedef struct { ...@@ -627,15 +644,16 @@ typedef struct {
} SRowInfo; } SRowInfo;
typedef struct SMergeTree { typedef struct SMergeTree {
int8_t backward; int8_t backward;
SRBTree rbt; SRBTree rbt;
SArray *pIterList; SArray *pIterList;
struct SLDataIter *pIter; SLDataIter *pIter;
} SMergeTree; } SMergeTree;
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange); int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t uid,
void tMergeTreeAddIter(SMergeTree *pMTree, struct SLDataIter *pIter); STimeWindow *pTimeWindow, SVersionRange *pVerRange);
bool tMergeTreeNext(SMergeTree *pMTree); void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
TSDBROW tMergeTreeGetRow(SMergeTree *pMTree); TSDBROW tMergeTreeGetRow(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree); void tMergeTreeClose(SMergeTree *pMTree);
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 0c27b07c
...@@ -71,7 +71,7 @@ typedef struct { ...@@ -71,7 +71,7 @@ typedef struct {
SDataIter *pIter; SDataIter *pIter;
SRBTree rbt; SRBTree rbt;
SDataIter dataIter; SDataIter dataIter;
SDataIter aDataIter[TSDB_MAX_LAST_FILE]; SDataIter aDataIter[TSDB_MAX_SST_FILE];
int8_t toLastOnly; int8_t toLastOnly;
}; };
struct { struct {
...@@ -92,9 +92,6 @@ typedef struct { ...@@ -92,9 +92,6 @@ typedef struct {
SArray *aDelData; // SArray<SDelData> SArray *aDelData; // SArray<SDelData>
} SCommitter; } SCommitter;
extern int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *aSstBlk,
SBlockData *pBlockData); // todo
static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter); static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter);
static int32_t tsdbCommitData(SCommitter *pCommitter); static int32_t tsdbCommitData(SCommitter *pCommitter);
static int32_t tsdbCommitDel(SCommitter *pCommitter); static int32_t tsdbCommitDel(SCommitter *pCommitter);
...@@ -445,7 +442,7 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) { ...@@ -445,7 +442,7 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) {
pIter->iSstBlk = 0; pIter->iSstBlk = 0;
SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, 0); SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, 0);
code = tsdbReadSstBlockEx(pCommitter->dReader.pReader, iSst, pSstBlk, &pIter->bData); code = tsdbReadSstBlock(pCommitter->dReader.pReader, iSst, pSstBlk, &pIter->bData);
if (code) goto _err; if (code) goto _err;
pIter->iRow = 0; pIter->iRow = 0;
...@@ -760,7 +757,7 @@ static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) { ...@@ -760,7 +757,7 @@ static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) {
pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows; pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows; pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression; pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
pCommitter->maxLast = TSDB_DEFAULT_LAST_FILE; // TODO: make it as a config pCommitter->maxLast = TSDB_DEFAULT_SST_FILE; // TODO: make it as a config
pCommitter->aTbDataP = tsdbMemTableGetTbDataArray(pTsdb->imem); pCommitter->aTbDataP = tsdbMemTableGetTbDataArray(pTsdb->imem);
if (pCommitter->aTbDataP == NULL) { if (pCommitter->aTbDataP == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
...@@ -790,7 +787,7 @@ static int32_t tsdbCommitDataStart(SCommitter *pCommitter) { ...@@ -790,7 +787,7 @@ static int32_t tsdbCommitDataStart(SCommitter *pCommitter) {
if (code) goto _exit; if (code) goto _exit;
// merger // merger
for (int32_t iSst = 0; iSst < TSDB_MAX_LAST_FILE; iSst++) { for (int32_t iSst = 0; iSst < TSDB_MAX_SST_FILE; iSst++) {
SDataIter *pIter = &pCommitter->aDataIter[iSst]; SDataIter *pIter = &pCommitter->aDataIter[iSst];
pIter->aSstBlk = taosArrayInit(0, sizeof(SSstBlk)); pIter->aSstBlk = taosArrayInit(0, sizeof(SSstBlk));
if (pIter->aSstBlk == NULL) { if (pIter->aSstBlk == NULL) {
...@@ -832,7 +829,7 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) { ...@@ -832,7 +829,7 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) {
tBlockDataDestroy(&pCommitter->dReader.bData, 1); tBlockDataDestroy(&pCommitter->dReader.bData, 1);
// merger // merger
for (int32_t iSst = 0; iSst < TSDB_MAX_LAST_FILE; iSst++) { for (int32_t iSst = 0; iSst < TSDB_MAX_SST_FILE; iSst++) {
SDataIter *pIter = &pCommitter->aDataIter[iSst]; SDataIter *pIter = &pCommitter->aDataIter[iSst];
taosArrayDestroy(pIter->aSstBlk); taosArrayDestroy(pIter->aSstBlk);
tBlockDataDestroy(&pIter->bData, 1); tBlockDataDestroy(&pIter->bData, 1);
...@@ -1059,7 +1056,7 @@ static int32_t tsdbNextCommitRow(SCommitter *pCommitter) { ...@@ -1059,7 +1056,7 @@ static int32_t tsdbNextCommitRow(SCommitter *pCommitter) {
if (pIter->iSstBlk < taosArrayGetSize(pIter->aSstBlk)) { if (pIter->iSstBlk < taosArrayGetSize(pIter->aSstBlk)) {
SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, pIter->iSstBlk); SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, pIter->iSstBlk);
code = tsdbReadSstBlockEx(pCommitter->dReader.pReader, pIter->iSst, pSstBlk, &pIter->bData); code = tsdbReadSstBlock(pCommitter->dReader.pReader, pIter->iSst, pSstBlk, &pIter->bData);
if (code) goto _exit; if (code) goto _exit;
pIter->iRow = 0; pIter->iRow = 0;
......
source/dnode/vnode/src/tsdb/tsdbFS.c
浏览文件 @ 0c27b07c
...@@ -21,6 +21,9 @@ static int32_t tsdbEncodeFS(uint8_t *p, STsdbFS *pFS) { ...@@ -21,6 +21,9 @@ static int32_t tsdbEncodeFS(uint8_t *p, STsdbFS *pFS) {
int8_t hasDel = pFS->pDelFile ? 1 : 0; int8_t hasDel = pFS->pDelFile ? 1 : 0;
uint32_t nSet = taosArrayGetSize(pFS->aDFileSet); uint32_t nSet = taosArrayGetSize(pFS->aDFileSet);
// version
n += tPutI8(p ? p + n : p, 0);
// SDelFile // SDelFile
n += tPutI8(p ? p + n : p, hasDel); n += tPutI8(p ? p + n : p, hasDel);
if (hasDel) { if (hasDel) {
...@@ -292,7 +295,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) { ...@@ -292,7 +295,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size != pSet->pHeadF->size) { if (size != LOGIC_TO_FILE_SIZE(pSet->pHeadF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} }
...@@ -303,10 +306,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) { ...@@ -303,10 +306,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size < pSet->pDataF->size) { if (size < LOGIC_TO_FILE_SIZE(pSet->pDataF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} else if (size > pSet->pDataF->size) { } else if (size > LOGIC_TO_FILE_SIZE(pSet->pDataF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE); code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE);
if (code) goto _err; if (code) goto _err;
} }
...@@ -317,10 +320,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) { ...@@ -317,10 +320,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size < pSet->pSmaF->size) { if (size < LOGIC_TO_FILE_SIZE(pSet->pSmaF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} else if (size > pSet->pSmaF->size) { } else if (size > LOGIC_TO_FILE_SIZE(pSet->pSmaF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE); code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
if (code) goto _err; if (code) goto _err;
} }
...@@ -332,7 +335,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) { ...@@ -332,7 +335,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size != pSet->aSstF[iSst]->size) { if (size != LOGIC_TO_FILE_SIZE(pSet->aSstF[iSst]->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} }
...@@ -364,10 +367,12 @@ static int32_t tsdbRecoverFS(STsdb *pTsdb, uint8_t *pData, int64_t nData) { ...@@ -364,10 +367,12 @@ static int32_t tsdbRecoverFS(STsdb *pTsdb, uint8_t *pData, int64_t nData) {
int32_t code = 0; int32_t code = 0;
int8_t hasDel; int8_t hasDel;
uint32_t nSet; uint32_t nSet;
int32_t n; int32_t n = 0;
// version
n += tGetI8(pData + n, NULL);
// SDelFile // SDelFile
n = 0;
n += tGetI8(pData + n, &hasDel); n += tGetI8(pData + n, &hasDel);
if (hasDel) { if (hasDel) {
pTsdb->fs.pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile)); pTsdb->fs.pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
......
source/dnode/vnode/src/tsdb/tsdbFile.c
浏览文件 @ 0c27b07c
...@@ -148,7 +148,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) { ...@@ -148,7 +148,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
} }
// ftruncate // ftruncate
if (taosFtruncateFile(pFD, size) < 0) { if (taosFtruncateFile(pFD, LOGIC_TO_FILE_SIZE(size, TSDB_DEFAULT_PAGE_SIZE)) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
......
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
浏览文件 @ 0c27b07c
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#include "tsdb.h" #include "tsdb.h"
// SLDataIter ================================================= // SLDataIter =================================================
typedef struct SLDataIter { struct SLDataIter {
SRBTreeNode node; SRBTreeNode node;
SSstBlk *pSstBlk; SSstBlk *pSstBlk;
SDataFReader *pReader; SDataFReader *pReader;
...@@ -31,13 +31,11 @@ typedef struct SLDataIter { ...@@ -31,13 +31,11 @@ typedef struct SLDataIter {
uint64_t uid; uint64_t uid;
STimeWindow timeWindow; STimeWindow timeWindow;
SVersionRange verRange; SVersionRange verRange;
} SLDataIter; };
static SBlockData* getCurrentBlock(SLDataIter* pIter) { static SBlockData *getCurrentBlock(SLDataIter *pIter) { return &pIter->bData[pIter->loadIndex]; }
return &pIter->bData[pIter->loadIndex];
}
static SBlockData* getNextBlock(SLDataIter* pIter) { static SBlockData *getNextBlock(SLDataIter *pIter) {
pIter->loadIndex ^= 1; pIter->loadIndex ^= 1;
return getCurrentBlock(pIter); return getCurrentBlock(pIter);
} }
...@@ -116,8 +114,6 @@ void tLDataIterClose(SLDataIter *pIter) { ...@@ -116,8 +114,6 @@ void tLDataIterClose(SLDataIter *pIter) {
taosMemoryFree(pIter); taosMemoryFree(pIter);
} }
extern int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData);
void tLDataIterNextBlock(SLDataIter *pIter) { void tLDataIterNextBlock(SLDataIter *pIter) {
int32_t step = pIter->backward ? -1 : 1; int32_t step = pIter->backward ? -1 : 1;
pIter->iSstBlk += step; pIter->iSstBlk += step;
...@@ -150,9 +146,9 @@ void tLDataIterNextBlock(SLDataIter *pIter) { ...@@ -150,9 +146,9 @@ void tLDataIterNextBlock(SLDataIter *pIter) {
static void findNextValidRow(SLDataIter *pIter) { static void findNextValidRow(SLDataIter *pIter) {
int32_t step = pIter->backward ? -1 : 1; int32_t step = pIter->backward ? -1 : 1;
bool hasVal = false; bool hasVal = false;
int32_t i = pIter->iRow; int32_t i = pIter->iRow;
SBlockData* pBlockData = getCurrentBlock(pIter); SBlockData *pBlockData = getCurrentBlock(pIter);
for (; i < pBlockData->nRow && i >= 0; i += step) { for (; i < pBlockData->nRow && i >= 0; i += step) {
if (pBlockData->aUid != NULL) { if (pBlockData->aUid != NULL) {
...@@ -220,12 +216,12 @@ bool tLDataIterNextRow(SLDataIter *pIter) { ...@@ -220,12 +216,12 @@ bool tLDataIterNextRow(SLDataIter *pIter) {
return false; return false;
} }
int32_t iBlockL = pIter->iSstBlk; int32_t iBlockL = pIter->iSstBlk;
SBlockData* pBlockData = getCurrentBlock(pIter); SBlockData *pBlockData = getCurrentBlock(pIter);
if (pBlockData->nRow == 0 && pIter->pSstBlk != NULL) { // current block not loaded yet if (pBlockData->nRow == 0 && pIter->pSstBlk != NULL) { // current block not loaded yet
pBlockData = getNextBlock(pIter); pBlockData = getNextBlock(pIter);
code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData); code = tsdbReadSstBlock(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _exit; goto _exit;
} }
...@@ -249,7 +245,7 @@ bool tLDataIterNextRow(SLDataIter *pIter) { ...@@ -249,7 +245,7 @@ bool tLDataIterNextRow(SLDataIter *pIter) {
if (iBlockL != pIter->iSstBlk) { if (iBlockL != pIter->iSstBlk) {
pBlockData = getNextBlock(pIter); pBlockData = getNextBlock(pIter);
code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData); code = tsdbReadSstBlock(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
if (code) { if (code) {
goto _exit; goto _exit;
} }
...@@ -306,7 +302,7 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead ...@@ -306,7 +302,7 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead
tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn); tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);
int32_t code = TSDB_CODE_OUT_OF_MEMORY; int32_t code = TSDB_CODE_OUT_OF_MEMORY;
struct SLDataIter *pIterList[TSDB_DEFAULT_LAST_FILE] = {0}; struct SLDataIter *pIterList[TSDB_DEFAULT_SST_FILE] = {0};
for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file
code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange); code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
......
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
浏览文件 @ 0c27b07c
...@@ -15,741 +15,944 @@ ...@@ -15,741 +15,944 @@
#include "tsdb.h" #include "tsdb.h"
// SDelFWriter ==================================================== // =============== PAGE-WISE FILE ===============
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) { static int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **ppFD) {
int32_t code = 0; int32_t code = 0;
char fname[TSDB_FILENAME_LEN]; STsdbFD *pFD;
char hdr[TSDB_FHDR_SIZE] = {0};
SDelFWriter *pDelFWriter;
int64_t n;
// alloc *ppFD = NULL;
pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
if (pDelFWriter == NULL) { pFD = (STsdbFD *)taosMemoryCalloc(1, sizeof(*pFD) + strlen(path) + 1);
if (pFD == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _exit;
} }
pDelFWriter->pTsdb = pTsdb;
pDelFWriter->fDel = *pFile;
tsdbDelFileName(pTsdb, pFile, fname); pFD->path = (char *)&pFD[1];
pDelFWriter->pWriteH = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE); strcpy(pFD->path, path);
if (pDelFWriter->pWriteH == NULL) { pFD->szPage = szPage;
pFD->flag = flag;
pFD->pFD = taosOpenFile(path, flag);
if (pFD->pFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} }
pFD->szPage = szPage;
// update header pFD->pgno = 0;
n = taosWriteFile(pDelFWriter->pWriteH, &hdr, TSDB_FHDR_SIZE); pFD->pBuf = taosMemoryCalloc(1, szPage);
if (n < 0) { if (pFD->pBuf == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pFD);
goto _exit;
}
if (taosStatFile(path, &pFD->szFile, NULL) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} }
ASSERT(pFD->szFile % szPage == 0);
pFD->szFile = pFD->szFile / szPage;
*ppFD = pFD;
pDelFWriter->fDel.size = TSDB_FHDR_SIZE; _exit:
pDelFWriter->fDel.offset = 0;
*ppWriter = pDelFWriter;
return code; return code;
}
_err: static void tsdbCloseFile(STsdbFD **ppFD) {
tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); STsdbFD *pFD = *ppFD;
*ppWriter = NULL; taosMemoryFree(pFD->pBuf);
taosCloseFile(&pFD->pFD);
taosMemoryFree(pFD);
*ppFD = NULL;
}
static int32_t tsdbWriteFilePage(STsdbFD *pFD) {
int32_t code = 0;
if (pFD->pgno > 0) {
int64_t n = taosLSeekFile(pFD->pFD, PAGE_OFFSET(pFD->pgno, pFD->szPage), SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
taosCalcChecksumAppend(0, pFD->pBuf, pFD->szPage);
n = taosWriteFile(pFD->pFD, pFD->pBuf, pFD->szPage);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
if (pFD->szFile < pFD->pgno) {
pFD->szFile = pFD->pgno;
}
}
pFD->pgno = 0;
_exit:
return code; return code;
} }
int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) { static int32_t tsdbReadFilePage(STsdbFD *pFD, int64_t pgno) {
int32_t code = 0; int32_t code = 0;
SDelFWriter *pWriter = *ppWriter;
STsdb *pTsdb = pWriter->pTsdb;
// sync ASSERT(pgno <= pFD->szFile);
if (sync && taosFsyncFile(pWriter->pWriteH) < 0) {
// seek
int64_t offset = PAGE_OFFSET(pgno, pFD->szPage);
int64_t n = taosLSeekFile(pFD->pFD, offset, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} }
// close // read
if (taosCloseFile(&pWriter->pWriteH) < 0) { n = taosReadFile(pFD->pFD, pFD->pBuf, pFD->szPage);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _exit;
} else if (n < pFD->szPage) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
} }
for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) { // check
tFree(pWriter->aBuf[iBuf]); if (!taosCheckChecksumWhole(pFD->pBuf, pFD->szPage)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
} }
taosMemoryFree(pWriter);
*ppWriter = NULL; pFD->pgno = pgno;
return code;
_err: _exit:
tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) { static int32_t tsdbWriteFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
int32_t code = 0; int32_t code = 0;
int64_t size; int64_t fOffset = LOGIC_TO_FILE_OFFSET(offset, pFD->szPage);
int64_t n; int64_t pgno = OFFSET_PGNO(fOffset, pFD->szPage);
int64_t bOffset = fOffset % pFD->szPage;
int64_t n = 0;
do {
if (pFD->pgno != pgno) {
code = tsdbWriteFilePage(pFD);
if (code) goto _exit;
if (pgno <= pFD->szFile) {
code = tsdbReadFilePage(pFD, pgno);
if (code) goto _exit;
} else {
pFD->pgno = pgno;
}
}
// prepare int64_t nWrite = TMIN(PAGE_CONTENT_SIZE(pFD->szPage) - bOffset, size - n);
size = sizeof(uint32_t); memcpy(pFD->pBuf + bOffset, pBuf + n, nWrite);
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
}
size += sizeof(TSCKSUM);
// alloc pgno++;
code = tRealloc(&pWriter->aBuf[0], size); bOffset = 0;
if (code) goto _err; n += nWrite;
} while (n < size);
// build _exit:
n = 0; return code;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); }
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size); static int32_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t size) {
int32_t code = 0;
int64_t n = 0;
int64_t fOffset = LOGIC_TO_FILE_OFFSET(offset, pFD->szPage);
int64_t pgno = OFFSET_PGNO(fOffset, pFD->szPage);
int32_t szPgCont = PAGE_CONTENT_SIZE(pFD->szPage);
int64_t bOffset = fOffset % pFD->szPage;
ASSERT(pgno && pgno <= pFD->szFile);
ASSERT(bOffset < szPgCont);
while (n < size) {
if (pFD->pgno != pgno) {
code = tsdbReadFilePage(pFD, pgno);
if (code) goto _exit;
}
// write int64_t nRead = TMIN(szPgCont - bOffset, size - n);
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size); memcpy(pBuf + n, pFD->pBuf + bOffset, nRead);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); n += nRead;
goto _err; pgno++;
bOffset = 0;
} }
ASSERT(n == size); _exit:
return code;
}
// update static int32_t tsdbFsyncFile(STsdbFD *pFD) {
pDelIdx->offset = pWriter->fDel.size; int32_t code = 0;
pDelIdx->size = size;
pWriter->fDel.size += size;
return code; code = tsdbWriteFilePage(pFD);
if (code) goto _exit;
_err: if (taosFsyncFile(pFD->pFD) < 0) {
tsdbError("vgId:%d, failed to write del data since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
_exit:
return code; return code;
} }
int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) { // SDataFWriter ====================================================
int32_t code = 0; int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
int64_t size; int32_t code = 0;
int64_t n; int32_t flag;
SDelIdx *pDelIdx; int64_t n;
int32_t szPage = TSDB_DEFAULT_PAGE_SIZE;
SDataFWriter *pWriter = NULL;
char fname[TSDB_FILENAME_LEN];
char hdr[TSDB_FHDR_SIZE] = {0};
// prepare // alloc
size = sizeof(uint32_t); pWriter = taosMemoryCalloc(1, sizeof(*pWriter));
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) { if (pWriter == NULL) {
size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx)); code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pWriter->pTsdb = pTsdb;
pWriter->wSet = (SDFileSet){.diskId = pSet->diskId,
.fid = pSet->fid,
.pHeadF = &pWriter->fHead,
.pDataF = &pWriter->fData,
.pSmaF = &pWriter->fSma,
.nSstF = pSet->nSstF};
pWriter->fHead = *pSet->pHeadF;
pWriter->fData = *pSet->pDataF;
pWriter->fSma = *pSet->pSmaF;
for (int8_t iSst = 0; iSst < pSet->nSstF; iSst++) {
pWriter->wSet.aSstF[iSst] = &pWriter->fSst[iSst];
pWriter->fSst[iSst] = *pSet->aSstF[iSst];
} }
size += sizeof(TSCKSUM);
// alloc // head
code = tRealloc(&pWriter->aBuf[0], size); flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname);
code = tsdbOpenFile(fname, szPage, flag, &pWriter->pHeadFD);
if (code) goto _err; if (code) goto _err;
// build code = tsdbWriteFile(pWriter->pHeadFD, 0, hdr, TSDB_FHDR_SIZE);
n = 0; if (code) goto _err;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); pWriter->fHead.size += TSDB_FHDR_SIZE;
for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx)); // data
if (pWriter->fData.size == 0) {
flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
} else {
flag = TD_FILE_READ | TD_FILE_WRITE;
}
tsdbDataFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fData, fname);
code = tsdbOpenFile(fname, szPage, flag, &pWriter->pDataFD);
if (code) goto _err;
if (pWriter->fData.size == 0) {
code = tsdbWriteFile(pWriter->pDataFD, 0, hdr, TSDB_FHDR_SIZE);
if (code) goto _err;
pWriter->fData.size += TSDB_FHDR_SIZE;
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size); // sma
if (pWriter->fSma.size == 0) {
flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
} else {
flag = TD_FILE_READ | TD_FILE_WRITE;
}
tsdbSmaFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSma, fname);
code = tsdbOpenFile(fname, szPage, flag, &pWriter->pSmaFD);
if (code) goto _err;
if (pWriter->fSma.size == 0) {
code = tsdbWriteFile(pWriter->pSmaFD, 0, hdr, TSDB_FHDR_SIZE);
if (code) goto _err;
// write pWriter->fSma.size += TSDB_FHDR_SIZE;
n = taosWriteFile(pWriter->pWriteH, pWriter->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
// update // sst
pWriter->fDel.offset = pWriter->fDel.size; ASSERT(pWriter->fSst[pSet->nSstF - 1].size == 0);
pWriter->fDel.size += size; flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
tsdbSstFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSst[pSet->nSstF - 1], fname);
code = tsdbOpenFile(fname, szPage, flag, &pWriter->pSstFD);
if (code) goto _err;
code = tsdbWriteFile(pWriter->pSstFD, 0, hdr, TSDB_FHDR_SIZE);
if (code) goto _err;
pWriter->fSst[pWriter->wSet.nSstF - 1].size += TSDB_FHDR_SIZE;
*ppWriter = pWriter;
return code; return code;
_err: _err:
tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, tsdb data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppWriter = NULL;
return code; return code;
} }
int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) { int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
int32_t code = 0; int32_t code = 0;
char hdr[TSDB_FHDR_SIZE]; STsdb *pTsdb = NULL;
int64_t size = TSDB_FHDR_SIZE;
int64_t n;
// build if (*ppWriter == NULL) goto _exit;
memset(hdr, 0, size);
tPutDelFile(hdr, &pWriter->fDel);
taosCalcChecksumAppend(0, hdr, size);
// seek pTsdb = (*ppWriter)->pTsdb;
if (taosLSeekFile(pWriter->pWriteH, 0, SEEK_SET) < 0) { if (sync) {
code = TAOS_SYSTEM_ERROR(errno); code = tsdbFsyncFile((*ppWriter)->pHeadFD);
goto _err; if (code) goto _err;
}
// write code = tsdbFsyncFile((*ppWriter)->pDataFD);
n = taosWriteFile(pWriter->pWriteH, hdr, size); if (code) goto _err;
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = tsdbFsyncFile((*ppWriter)->pSmaFD);
goto _err; if (code) goto _err;
code = tsdbFsyncFile((*ppWriter)->pSstFD);
if (code) goto _err;
} }
tsdbCloseFile(&(*ppWriter)->pHeadFD);
tsdbCloseFile(&(*ppWriter)->pDataFD);
tsdbCloseFile(&(*ppWriter)->pSmaFD);
tsdbCloseFile(&(*ppWriter)->pSstFD);
for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree((*ppWriter)->aBuf[iBuf]);
}
taosMemoryFree(*ppWriter);
_exit:
*ppWriter = NULL;
return code; return code;
_err: _err:
tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, data file writer close failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
// SDelFReader ==================================================== int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
struct SDelFReader { int32_t code = 0;
STsdb *pTsdb; int64_t n;
SDelFile fDel; char hdr[TSDB_FHDR_SIZE];
TdFilePtr pReadH;
uint8_t *aBuf[1];
};
int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) { // head ==============
int32_t code = 0; memset(hdr, 0, TSDB_FHDR_SIZE);
char fname[TSDB_FILENAME_LEN]; tPutHeadFile(hdr, &pWriter->fHead);
SDelFReader *pDelFReader; code = tsdbWriteFile(pWriter->pHeadFD, 0, hdr, TSDB_FHDR_SIZE);
int64_t n; if (code) goto _err;
// alloc // data ==============
pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader)); memset(hdr, 0, TSDB_FHDR_SIZE);
if (pDelFReader == NULL) { tPutDataFile(hdr, &pWriter->fData);
code = TSDB_CODE_OUT_OF_MEMORY; code = tsdbWriteFile(pWriter->pDataFD, 0, hdr, TSDB_FHDR_SIZE);
goto _err; if (code) goto _err;
}
// open impl // sma ==============
pDelFReader->pTsdb = pTsdb; memset(hdr, 0, TSDB_FHDR_SIZE);
pDelFReader->fDel = *pFile; tPutSmaFile(hdr, &pWriter->fSma);
code = tsdbWriteFile(pWriter->pSmaFD, 0, hdr, TSDB_FHDR_SIZE);
if (code) goto _err;
tsdbDelFileName(pTsdb, pFile, fname); // sst ==============
pDelFReader->pReadH = taosOpenFile(fname, TD_FILE_READ); memset(hdr, 0, TSDB_FHDR_SIZE);
if (pDelFReader->pReadH == NULL) { tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]);
code = TAOS_SYSTEM_ERROR(errno); code = tsdbWriteFile(pWriter->pSstFD, 0, hdr, TSDB_FHDR_SIZE);
taosMemoryFree(pDelFReader); if (code) goto _err;
goto _err;
}
_exit:
*ppReader = pDelFReader;
return code; return code;
_err: _err:
tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, update DFileSet header failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
*ppReader = NULL;
return code; return code;
} }
int32_t tsdbDelFReaderClose(SDelFReader **ppReader) { int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) {
int32_t code = 0; int32_t code = 0;
SDelFReader *pReader = *ppReader; SHeadFile *pHeadFile = &pWriter->fHead;
int64_t size;
int64_t n;
if (pReader) { // check
if (taosCloseFile(&pReader->pReadH) < 0) { if (taosArrayGetSize(aBlockIdx) == 0) {
code = TAOS_SYSTEM_ERROR(errno); pHeadFile->offset = pHeadFile->size;
goto _exit; goto _exit;
}
for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pReader->aBuf[iBuf]);
}
taosMemoryFree(pReader);
} }
*ppReader = NULL;
_exit:
return code;
}
int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) {
int32_t code = 0;
int64_t offset = pDelIdx->offset;
int64_t size = pDelIdx->size;
int64_t n;
taosArrayClear(aDelData);
// seek // prepare
if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) { size = 0;
code = TAOS_SYSTEM_ERROR(errno); for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
goto _err; size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
} }
// alloc // alloc
code = tRealloc(&pReader->aBuf[0], size); code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// read // build
n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// // decode
n = 0; n = 0;
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
uint32_t delimiter; n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
while (n < size - sizeof(TSCKSUM)) {
SDelData delData;
n += tGetDelData(pReader->aBuf[0] + n, &delData);
if (taosArrayPush(aDelData, &delData) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
} }
ASSERT(n == size);
// write
code = tsdbWriteFile(pWriter->pHeadFD, pHeadFile->size, pWriter->aBuf[0], size);
if (code) goto _err;
ASSERT(n == size - sizeof(TSCKSUM)); // update
pHeadFile->offset = pHeadFile->size;
pHeadFile->size += size;
_exit:
// tsdbTrace("vgId:%d write block idx, offset:%" PRId64 " size:%" PRId64 " nBlockIdx:%d",
// TD_VID(pWriter->pTsdb->pVnode),
// pHeadFile->offset, size, taosArrayGetSize(aBlockIdx));
return code; return code;
_err: _err:
tsdbError("vgId:%d, read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, write block idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) { int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) {
int32_t code = 0; int32_t code = 0;
int32_t n; SHeadFile *pHeadFile = &pWriter->fHead;
int64_t offset = pReader->fDel.offset; int64_t size;
int64_t size = pReader->fDel.size - offset; int64_t n;
taosArrayClear(aDelIdx);
// seek ASSERT(mBlock->nItem > 0);
if (taosLSeekFile(pReader->pReadH, offset, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// alloc // alloc
code = tRealloc(&pReader->aBuf[0], size); size = tPutMapData(NULL, mBlock);
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// read // build
n = taosReadFile(pReader->pReadH, pReader->aBuf[0], size); n = tPutMapData(pWriter->aBuf[0], mBlock);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// decode
n = 0;
uint32_t delimiter;
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) {
SDelIdx delIdx;
n += tGetDelIdx(pReader->aBuf[0] + n, &delIdx);
if (taosArrayPush(aDelIdx, &delIdx) == NULL) { // write
code = TSDB_CODE_OUT_OF_MEMORY; code = tsdbWriteFile(pWriter->pHeadFD, pHeadFile->size, pWriter->aBuf[0], size);
goto _err; if (code) goto _err;
}
}
ASSERT(n == size - sizeof(TSCKSUM)); // update
pBlockIdx->offset = pHeadFile->size;
pBlockIdx->size = size;
pHeadFile->size += size;
tsdbTrace("vgId:%d, write block, file ID:%d commit ID:%d suid:%" PRId64 " uid:%" PRId64 " offset:%" PRId64
" size:%" PRId64 " nItem:%d",
TD_VID(pWriter->pTsdb->pVnode), pWriter->wSet.fid, pHeadFile->commitID, pBlockIdx->suid, pBlockIdx->uid,
pBlockIdx->offset, pBlockIdx->size, mBlock->nItem);
return code; return code;
_err: _err:
tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
// SDataFReader ==================================================== int32_t tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk) {
int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) { int32_t code = 0;
int32_t code = 0; SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1];
SDataFReader *pReader; int64_t size;
char fname[TSDB_FILENAME_LEN]; int64_t n;
// alloc // check
pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader)); if (taosArrayGetSize(aSstBlk) == 0) {
if (pReader == NULL) { pSstFile->offset = pSstFile->size;
code = TSDB_CODE_OUT_OF_MEMORY; goto _exit;
goto _err;
} }
pReader->pTsdb = pTsdb;
pReader->pSet = pSet;
// open impl // size
// head size = 0;
tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname); for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
pReader->pHeadFD = taosOpenFile(fname, TD_FILE_READ); size += tPutSstBlk(NULL, taosArrayGet(aSstBlk, iBlockL));
if (pReader->pHeadFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
// data // alloc
tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname); code = tRealloc(&pWriter->aBuf[0], size);
pReader->pDataFD = taosOpenFile(fname, TD_FILE_READ); if (code) goto _err;
if (pReader->pDataFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// sma // encode
tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname); n = 0;
pReader->pSmaFD = taosOpenFile(fname, TD_FILE_READ); for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
if (pReader->pSmaFD == NULL) { n += tPutSstBlk(pWriter->aBuf[0] + n, taosArrayGet(aSstBlk, iBlockL));
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
// sst // write
for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) { code = tsdbWriteFile(pWriter->pSstFD, pSstFile->size, pWriter->aBuf[0], size);
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname); if (code) goto _err;
pReader->aLastFD[iSst] = taosOpenFile(fname, TD_FILE_READ);
if (pReader->aLastFD[iSst] == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
*ppReader = pReader; // update
pSstFile->offset = pSstFile->size;
pSstFile->size += size;
_exit:
tsdbTrace("vgId:%d tsdb write sst block, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
pSstFile->offset, size);
return code; return code;
_err: _err:
tsdbError("vgId:%d, tsdb data file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d tsdb write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
*ppReader = NULL;
return code; return code;
} }
int32_t tsdbDataFReaderClose(SDataFReader **ppReader) { static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, SSmaInfo *pSmaInfo) {
int32_t code = 0; int32_t code = 0;
if (*ppReader == NULL) goto _exit;
// head pSmaInfo->offset = 0;
if (taosCloseFile(&(*ppReader)->pHeadFD) < 0) { pSmaInfo->size = 0;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// data // encode
if (taosCloseFile(&(*ppReader)->pDataFD) < 0) { for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
code = TAOS_SYSTEM_ERROR(errno); SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
goto _err;
}
// sma if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type)) continue;
if (taosCloseFile(&(*ppReader)->pSmaFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// sst SColumnDataAgg sma;
for (int32_t iSst = 0; iSst < (*ppReader)->pSet->nSstF; iSst++) { tsdbCalcColDataSMA(pColData, &sma);
if (taosCloseFile(&(*ppReader)->aLastFD[iSst]) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma));
goto _err; if (code) goto _err;
} pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma);
} }
for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) { // write
tFree((*ppReader)->aBuf[iBuf]); if (pSmaInfo->size) {
code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size);
if (code) goto _err;
code = tsdbWriteFile(pWriter->pSmaFD, pWriter->fSma.size, pWriter->aBuf[0], pSmaInfo->size);
if (code) goto _err;
pSmaInfo->offset = pWriter->fSma.size;
pWriter->fSma.size += pSmaInfo->size;
} }
taosMemoryFree(*ppReader);
_exit:
*ppReader = NULL;
return code; return code;
_err: _err:
tsdbError("vgId:%d, data file reader close failed since %s", TD_VID((*ppReader)->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d tsdb write block sma failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) { int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
int32_t code = 0; int8_t cmprAlg, int8_t toLast) {
int64_t offset = pReader->pSet->pHeadF->offset; int32_t code = 0;
int64_t size = pReader->pSet->pHeadF->size - offset;
int64_t n;
uint32_t delimiter;
taosArrayClear(aBlockIdx); ASSERT(pBlockData->nRow > 0);
if (size == 0) {
goto _exit; if (toLast) {
pBlkInfo->offset = pWriter->fSst[pWriter->wSet.nSstF - 1].size;
} else {
pBlkInfo->offset = pWriter->fData.size;
} }
pBlkInfo->szBlock = 0;
pBlkInfo->szKey = 0;
// alloc int32_t aBufN[4] = {0};
code = tRealloc(&pReader->aBuf[0], size); code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN);
if (code) goto _err; if (code) goto _err;
// seek // write =================
if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) { STsdbFD *pFD = toLast ? pWriter->pSstFD : pWriter->pDataFD;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// read pBlkInfo->szKey = aBufN[3] + aBufN[2];
n = taosReadFile(pReader->pHeadFD, pReader->aBuf[0], size); pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// check int64_t offset = pBlkInfo->offset;
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { code = tsdbWriteFile(pFD, offset, pWriter->aBuf[3], aBufN[3]);
code = TSDB_CODE_FILE_CORRUPTED; if (code) goto _err;
goto _err; offset += aBufN[3];
}
// decode code = tsdbWriteFile(pFD, offset, pWriter->aBuf[2], aBufN[2]);
n = 0; if (code) goto _err;
n = tGetU32(pReader->aBuf[0] + n, &delimiter); offset += aBufN[2];
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) { if (aBufN[1]) {
SBlockIdx blockIdx; code = tsdbWriteFile(pFD, offset, pWriter->aBuf[1], aBufN[1]);
n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx); if (code) goto _err;
offset += aBufN[1];
}
if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) { if (aBufN[0]) {
code = TSDB_CODE_OUT_OF_MEMORY; code = tsdbWriteFile(pFD, offset, pWriter->aBuf[0], aBufN[0]);
goto _err; if (code) goto _err;
} }
// update info
if (toLast) {
pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock;
} else {
pWriter->fData.size += pBlkInfo->szBlock;
} }
ASSERT(n + sizeof(TSCKSUM) == size); // ================= SMA ====================
if (pSmaInfo) {
code = tsdbWriteBlockSma(pWriter, pBlockData, pSmaInfo);
if (code) goto _err;
}
_exit: _exit:
tsdbTrace("vgId:%d tsdb write block data, suid:%" PRId64 " uid:%" PRId64 " nRow:%d, offset:%" PRId64 " size:%d",
TD_VID(pWriter->pTsdb->pVnode), pBlockData->suid, pBlockData->uid, pBlockData->nRow, pBlkInfo->offset,
pBlkInfo->szBlock);
return code; return code;
_err: _err:
tsdbError("vgId:%d, read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d tsdb write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbReadSstBlk(SDataFReader *pReader, int32_t iSst, SArray *aSstBlk) { int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pReader->pSet->aSstF[iSst]->offset; int64_t n;
int64_t size = pReader->pSet->aSstF[iSst]->size - offset; int64_t size;
int64_t n; TdFilePtr pOutFD = NULL; // TODO
uint32_t delimiter; TdFilePtr PInFD = NULL; // TODO
char fNameFrom[TSDB_FILENAME_LEN];
taosArrayClear(aSstBlk); char fNameTo[TSDB_FILENAME_LEN];
if (size == 0) {
goto _exit;
}
// alloc // head
code = tRealloc(&pReader->aBuf[0], size); tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom);
if (code) goto _err; tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
// seek pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (taosLSeekFile(pReader->aLastFD[iSst], offset, SEEK_SET) < 0) { if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// read PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
n = taosReadFile(pReader->aLastFD[iSst], pReader->aBuf[0], size); if (PInFD == NULL) {
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
// check n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size);
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { if (n < 0) {
code = TSDB_CODE_FILE_CORRUPTED; code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// decode // data
n = 0; tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom);
n = tGetU32(pReader->aBuf[0] + n, &delimiter); tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo);
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) {
SSstBlk blockl;
n += tGetSstBlk(pReader->aBuf[0] + n, &blockl);
if (taosArrayPush(aSstBlk, &blockl) == NULL) { pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
code = TSDB_CODE_OUT_OF_MEMORY; if (pOutFD == NULL) {
goto _err; code = TAOS_SYSTEM_ERROR(errno);
} goto _err;
} }
ASSERT(n + sizeof(TSCKSUM) == size); PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
_exit:
return code;
_err:
tsdbError("vgId:%d read blockl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) {
int32_t code = 0;
int64_t offset = pBlockIdx->offset;
int64_t size = pBlockIdx->size;
int64_t n;
int64_t tn;
// alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// seek
if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// read n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size);
n = taosReadFile(pReader->pHeadFD, pReader->aBuf[0], size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// check // sst
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { tsdbSstFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->aSstF[0], fNameFrom);
code = TSDB_CODE_FILE_CORRUPTED; 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) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// decode PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
n = 0; if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
uint32_t delimiter;
n += tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
tn = tGetMapData(pReader->aBuf[0] + n, mBlock);
if (tn < 0) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
n += tn;
ASSERT(n + sizeof(TSCKSUM) == size);
return code;
_err:
tsdbError("vgId:%d, read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aColumnDataAgg) {
int32_t code = 0;
SSmaInfo *pSmaInfo = &pDataBlk->smaInfo;
ASSERT(pSmaInfo->size > 0);
taosArrayClear(aColumnDataAgg);
// alloc
int32_t size = pSmaInfo->size + sizeof(TSCKSUM);
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// seek n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
int64_t n = taosLSeekFile(pReader->pSmaFD, pSmaInfo->offset, SEEK_SET);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} else if (n < pSmaInfo->offset) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// read // sma
n = taosReadFile(pReader->pSmaFD, pReader->aBuf[0], size); tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom);
if (n < 0) { tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
// check PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { if (PInFD == NULL) {
code = TSDB_CODE_FILE_CORRUPTED; code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// decode n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
n = 0; if (n < 0) {
while (n < pSmaInfo->size) { code = TAOS_SYSTEM_ERROR(errno);
SColumnDataAgg sma; goto _err;
n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
return code; return code;
_err: _err:
tsdbError("vgId:%d tsdb read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, int8_t fromLast, // SDataFReader ====================================================
SBlockData *pBlockData) { int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
int32_t code = 0; int32_t code = 0;
SDataFReader *pReader;
tBlockDataClear(pBlockData); int32_t szPage = TSDB_DEFAULT_PAGE_SIZE;
char fname[TSDB_FILENAME_LEN];
TdFilePtr pFD = fromLast ? pReader->aLastFD[0] : pReader->pDataFD; // (todo) // alloc
pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader));
if (pReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pReader->pTsdb = pTsdb;
pReader->pSet = pSet;
// uid + version + tskey // head
code = tsdbReadAndCheck(pFD, pBlkInfo->offset, &pReader->aBuf[0], pBlkInfo->szKey, 1); tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pHeadFD);
if (code) goto _err; if (code) goto _err;
SDiskDataHdr hdr;
uint8_t *p = pReader->aBuf[0] + tGetDiskDataHdr(pReader->aBuf[0], &hdr);
ASSERT(hdr.delimiter == TSDB_FILE_DLMT); // data
tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pDataFD);
if (code) goto _err;
// sma
tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->pSmaFD);
if (code) goto _err;
// sst
for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname);
code = tsdbOpenFile(fname, szPage, TD_FILE_READ, &pReader->aSstFD[iSst]);
if (code) goto _err;
}
*ppReader = pReader;
return code;
_err:
tsdbError("vgId:%d, tsdb data file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppReader = NULL;
return code;
}
int32_t tsdbDataFReaderClose(SDataFReader **ppReader) {
int32_t code = 0;
if (*ppReader == NULL) return code;
// head
tsdbCloseFile(&(*ppReader)->pHeadFD);
// data
tsdbCloseFile(&(*ppReader)->pDataFD);
// sma
tsdbCloseFile(&(*ppReader)->pSmaFD);
// sst
for (int32_t iSst = 0; iSst < TSDB_MAX_SST_FILE; iSst++) {
if ((*ppReader)->aSstFD[iSst]) {
tsdbCloseFile(&(*ppReader)->aSstFD[iSst]);
}
}
for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree((*ppReader)->aBuf[iBuf]);
}
taosMemoryFree(*ppReader);
*ppReader = NULL;
return code;
_err:
tsdbError("vgId:%d, data file reader close failed since %s", TD_VID((*ppReader)->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) {
int32_t code = 0;
SHeadFile *pHeadFile = pReader->pSet->pHeadF;
int64_t offset = pHeadFile->offset;
int64_t size = pHeadFile->size - offset;
taosArrayClear(aBlockIdx);
if (size == 0) return code;
// alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read
code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size);
if (code) goto _err;
// decode
int64_t n = 0;
while (n < size) {
SBlockIdx blockIdx;
n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx);
if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
ASSERT(n == size);
return code;
_err:
tsdbError("vgId:%d, read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadSstBlk(SDataFReader *pReader, int32_t iSst, SArray *aSstBlk) {
int32_t code = 0;
SSstFile *pSstFile = pReader->pSet->aSstF[iSst];
int64_t offset = pSstFile->offset;
int64_t size = pSstFile->size - offset;
taosArrayClear(aSstBlk);
if (size == 0) return code;
// alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read
code = tsdbReadFile(pReader->aSstFD[iSst], offset, pReader->aBuf[0], size);
if (code) goto _err;
// decode
int64_t n = 0;
while (n < size) {
SSstBlk sstBlk;
n += tGetSstBlk(pReader->aBuf[0] + n, &sstBlk);
if (taosArrayPush(aSstBlk, &sstBlk) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
ASSERT(n == size);
return code;
_err:
tsdbError("vgId:%d read sst blk failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) {
int32_t code = 0;
int64_t offset = pBlockIdx->offset;
int64_t size = pBlockIdx->size;
// alloc
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
// read
code = tsdbReadFile(pReader->pHeadFD, offset, pReader->aBuf[0], size);
if (code) goto _err;
// decode
int64_t n = tGetMapData(pReader->aBuf[0], mBlock);
if (n < 0) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
ASSERT(n == size);
return code;
_err:
tsdbError("vgId:%d, read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aColumnDataAgg) {
int32_t code = 0;
SSmaInfo *pSmaInfo = &pDataBlk->smaInfo;
ASSERT(pSmaInfo->size > 0);
taosArrayClear(aColumnDataAgg);
// alloc
code = tRealloc(&pReader->aBuf[0], pSmaInfo->size);
if (code) goto _err;
// read
code = tsdbReadFile(pReader->pSmaFD, pSmaInfo->offset, pReader->aBuf[0], pSmaInfo->size);
if (code) goto _err;
// decode
int32_t n = 0;
while (n < pSmaInfo->size) {
SColumnDataAgg sma;
n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
ASSERT(n == pSmaInfo->size);
return code;
_err:
tsdbError("vgId:%d tsdb read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, SBlockData *pBlockData) {
int32_t code = 0;
tBlockDataClear(pBlockData);
STsdbFD *pFD = pReader->pDataFD;
// uid + version + tskey
code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey);
if (code) goto _err;
code = tsdbReadFile(pFD, pBlkInfo->offset, pReader->aBuf[0], pBlkInfo->szKey);
if (code) goto _err;
SDiskDataHdr hdr;
uint8_t *p = pReader->aBuf[0] + tGetDiskDataHdr(pReader->aBuf[0], &hdr);
ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
ASSERT(pBlockData->suid == hdr.suid); ASSERT(pBlockData->suid == hdr.suid);
ASSERT(pBlockData->uid == hdr.uid); ASSERT(pBlockData->uid == hdr.uid);
...@@ -778,14 +981,18 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo ...@@ -778,14 +981,18 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo
if (code) goto _err; if (code) goto _err;
p += hdr.szKey; p += hdr.szKey;
ASSERT(p - pReader->aBuf[0] == pBlkInfo->szKey - sizeof(TSCKSUM)); ASSERT(p - pReader->aBuf[0] == pBlkInfo->szKey);
// read and decode columns // read and decode columns
if (taosArrayGetSize(pBlockData->aIdx) == 0) goto _exit; if (taosArrayGetSize(pBlockData->aIdx) == 0) goto _exit;
if (hdr.szBlkCol > 0) { if (hdr.szBlkCol > 0) {
int64_t offset = pBlkInfo->offset + pBlkInfo->szKey; int64_t offset = pBlkInfo->offset + pBlkInfo->szKey;
code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[0], hdr.szBlkCol + sizeof(TSCKSUM), 1);
code = tRealloc(&pReader->aBuf[0], hdr.szBlkCol);
if (code) goto _err;
code = tsdbReadFile(pFD, offset, pReader->aBuf[0], hdr.szBlkCol);
if (code) goto _err; if (code) goto _err;
} }
...@@ -823,10 +1030,13 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo ...@@ -823,10 +1030,13 @@ static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo
} }
} else { } else {
// decode from binary // decode from binary
int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + sizeof(TSCKSUM) + pBlockCol->offset; int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + pBlockCol->offset;
int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM); int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue;
code = tRealloc(&pReader->aBuf[1], size);
if (code) goto _err;
code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[1], size, 0); code = tsdbReadFile(pFD, offset, pReader->aBuf[1], size);
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]);
...@@ -846,7 +1056,7 @@ _err: ...@@ -846,7 +1056,7 @@ _err:
int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) { int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], 0, pBlockData); code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], pBlockData);
if (code) goto _err; if (code) goto _err;
if (pDataBlk->nSubBlock > 1) { if (pDataBlk->nSubBlock > 1) {
...@@ -864,7 +1074,7 @@ int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData ...@@ -864,7 +1074,7 @@ int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData
tBlockDataInitEx(&bData2, pBlockData); tBlockDataInitEx(&bData2, pBlockData);
for (int32_t iSubBlock = 1; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) { for (int32_t iSubBlock = 1; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[iSubBlock], 0, &bData1); code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[iSubBlock], &bData1);
if (code) { if (code) {
tBlockDataDestroy(&bData1, 1); tBlockDataDestroy(&bData1, 1);
tBlockDataDestroy(&bData2, 1); tBlockDataDestroy(&bData2, 1);
...@@ -900,327 +1110,139 @@ _err: ...@@ -900,327 +1110,139 @@ _err:
int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) { int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
code = tsdbReadBlockDataImpl(pReader, &pSstBlk->bInfo, 1, pBlockData); // alloc
code = tRealloc(&pReader->aBuf[0], pSstBlk->bInfo.szBlock);
if (code) goto _err; if (code) goto _err;
return code;
_err:
tsdbError("vgId:%d tsdb read last block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
int32_t code = 0;
// read // read
code = tsdbReadAndCheck(pReader->aLastFD[iSst], pSstBlk->bInfo.offset, &pReader->aBuf[0], pSstBlk->bInfo.szBlock, 0); code = tsdbReadFile(pReader->aSstFD[iSst], pSstBlk->bInfo.offset, pReader->aBuf[0], pSstBlk->bInfo.szBlock);
if (code) goto _exit; if (code) goto _err;
// decmpr // decmpr
code = tDecmprBlockData(pReader->aBuf[0], pSstBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]); code = tDecmprBlockData(pReader->aBuf[0], pSstBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]);
if (code) goto _exit; if (code) goto _err;
_exit: return code;
_err:
tsdbError("vgId:%d tsdb read sst block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
// SDataFWriter ==================================================== // SDelFWriter ====================================================
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) { int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
int32_t flag; char fname[TSDB_FILENAME_LEN];
int64_t n; uint8_t hdr[TSDB_FHDR_SIZE] = {0};
SDataFWriter *pWriter = NULL; SDelFWriter *pDelFWriter;
char fname[TSDB_FILENAME_LEN]; int64_t n;
char hdr[TSDB_FHDR_SIZE] = {0};
// alloc // alloc
pWriter = taosMemoryCalloc(1, sizeof(*pWriter)); pDelFWriter = (SDelFWriter *)taosMemoryCalloc(1, sizeof(*pDelFWriter));
if (pWriter == NULL) { if (pDelFWriter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
pWriter->pTsdb = pTsdb; pDelFWriter->pTsdb = pTsdb;
pWriter->wSet = (SDFileSet){ pDelFWriter->fDel = *pFile;
.diskId = pSet->diskId,
.fid = pSet->fid,
.pHeadF = &pWriter->fHead,
.pDataF = &pWriter->fData,
.pSmaF = &pWriter->fSma,
.nSstF = pSet->nSstF //
};
pWriter->fHead = *pSet->pHeadF;
pWriter->fData = *pSet->pDataF;
pWriter->fSma = *pSet->pSmaF;
for (int8_t iSst = 0; iSst < pSet->nSstF; iSst++) {
pWriter->wSet.aSstF[iSst] = &pWriter->fSst[iSst];
pWriter->fSst[iSst] = *pSet->aSstF[iSst];
}
// head tsdbDelFileName(pTsdb, pFile, fname);
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC; code =
tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname); tsdbOpenFile(fname, TSDB_DEFAULT_PAGE_SIZE, TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE, &pDelFWriter->pWriteH);
pWriter->pHeadFD = taosOpenFile(fname, flag); if (code) goto _err;
if (pWriter->pHeadFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE); // update header
if (n < 0) { code = tsdbWriteFile(pDelFWriter->pWriteH, 0, hdr, TSDB_FHDR_SIZE);
code = TAOS_SYSTEM_ERROR(errno); if (code) goto _err;
goto _err;
}
ASSERT(n == TSDB_FHDR_SIZE); pDelFWriter->fDel.size = TSDB_FHDR_SIZE;
pDelFWriter->fDel.offset = 0;
pWriter->fHead.size += TSDB_FHDR_SIZE; *ppWriter = pDelFWriter;
return code;
// data
if (pWriter->fData.size == 0) {
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
} else {
flag = TD_FILE_WRITE;
}
tsdbDataFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fData, fname);
pWriter->pDataFD = taosOpenFile(fname, flag);
if (pWriter->pDataFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (pWriter->fData.size == 0) {
n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pWriter->fData.size += TSDB_FHDR_SIZE;
} else {
n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_END);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
ASSERT(n == pWriter->fData.size);
}
// sma
if (pWriter->fSma.size == 0) {
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
} else {
flag = TD_FILE_WRITE;
}
tsdbSmaFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fSma, fname);
pWriter->pSmaFD = taosOpenFile(fname, flag);
if (pWriter->pSmaFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (pWriter->fSma.size == 0) {
n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pWriter->fSma.size += TSDB_FHDR_SIZE;
} else {
n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_END);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
ASSERT(n == pWriter->fSma.size);
}
// sst
ASSERT(pWriter->fSst[pSet->nSstF - 1].size == 0);
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
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);
goto _err;
}
n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
pWriter->fSst[pWriter->wSet.nSstF - 1].size += TSDB_FHDR_SIZE;
*ppWriter = pWriter;
return code;
_err: _err:
tsdbError("vgId:%d, tsdb data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, failed to open del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppWriter = NULL; *ppWriter = NULL;
return code; return code;
} }
int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) { int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync) {
int32_t code = 0; int32_t code = 0;
STsdb *pTsdb = NULL; SDelFWriter *pWriter = *ppWriter;
STsdb *pTsdb = pWriter->pTsdb;
if (*ppWriter == NULL) goto _exit;
pTsdb = (*ppWriter)->pTsdb; // sync
if (sync) { if (sync) {
if (taosFsyncFile((*ppWriter)->pHeadFD) < 0) { code = tsdbFsyncFile(pWriter->pWriteH);
code = TAOS_SYSTEM_ERROR(errno); if (code) goto _err;
goto _err;
}
if (taosFsyncFile((*ppWriter)->pDataFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile((*ppWriter)->pSmaFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile((*ppWriter)->pLastFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
if (taosCloseFile(&(*ppWriter)->pHeadFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosCloseFile(&(*ppWriter)->pDataFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
if (taosCloseFile(&(*ppWriter)->pSmaFD) < 0) { // close
code = TAOS_SYSTEM_ERROR(errno); tsdbCloseFile(&pWriter->pWriteH);
goto _err;
}
if (taosCloseFile(&(*ppWriter)->pLastFD) < 0) { for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t *); iBuf++) {
code = TAOS_SYSTEM_ERROR(errno); tFree(pWriter->aBuf[iBuf]);
goto _err;
} }
taosMemoryFree(pWriter);
for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree((*ppWriter)->aBuf[iBuf]);
}
taosMemoryFree(*ppWriter);
_exit:
*ppWriter = NULL; *ppWriter = NULL;
return code; return code;
_err: _err:
tsdbError("vgId:%d, data file writer close failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, failed to close del file writer since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) { int32_t tsdbWriteDelData(SDelFWriter *pWriter, SArray *aDelData, SDelIdx *pDelIdx) {
int32_t code = 0; int32_t code = 0;
int64_t size;
int64_t n; int64_t n;
char hdr[TSDB_FHDR_SIZE];
// head ==============
memset(hdr, 0, TSDB_FHDR_SIZE);
tPutHeadFile(hdr, &pWriter->fHead);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pHeadFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// data ==============
memset(hdr, 0, TSDB_FHDR_SIZE);
tPutDataFile(hdr, &pWriter->fData);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE); // prepare
if (n < 0) { size = 0;
code = TAOS_SYSTEM_ERROR(errno); for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
goto _err; size += tPutDelData(NULL, taosArrayGet(aDelData, iDelData));
} }
// sma ============== // alloc
memset(hdr, 0, TSDB_FHDR_SIZE); code = tRealloc(&pWriter->aBuf[0], size);
tPutSmaFile(hdr, &pWriter->fSma); if (code) goto _err;
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE); // build
if (n < 0) { n = 0;
code = TAOS_SYSTEM_ERROR(errno); for (int32_t iDelData = 0; iDelData < taosArrayGetSize(aDelData); iDelData++) {
goto _err; n += tPutDelData(pWriter->aBuf[0] + n, taosArrayGet(aDelData, iDelData));
} }
ASSERT(n == size);
// sst ============== // write
memset(hdr, 0, TSDB_FHDR_SIZE); code = tsdbWriteFile(pWriter->pWriteH, pWriter->fDel.size, pWriter->aBuf[0], size);
tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]); if (code) goto _err;
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE); // update
if (n < 0) { pDelIdx->offset = pWriter->fDel.size;
code = TAOS_SYSTEM_ERROR(errno); pDelIdx->size = size;
goto _err; pWriter->fDel.size += size;
}
return code; return code;
_err: _err:
tsdbError("vgId:%d, update DFileSet header failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, failed to write del data since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) { int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SArray *aDelIdx) {
int32_t code = 0; int32_t code = 0;
SHeadFile *pHeadFile = &pWriter->fHead; int64_t size;
int64_t size = 0; int64_t n;
int64_t n; SDelIdx *pDelIdx;
// check
if (taosArrayGetSize(aBlockIdx) == 0) {
pHeadFile->offset = pHeadFile->size;
goto _exit;
}
// prepare // prepare
size = sizeof(uint32_t); size = 0;
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) { for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx)); size += tPutDelIdx(NULL, taosArrayGet(aDelIdx, iDelIdx));
} }
size += sizeof(TSCKSUM);
// alloc // alloc
code = tRealloc(&pWriter->aBuf[0], size); code = tRealloc(&pWriter->aBuf[0], size);
...@@ -1228,383 +1250,170 @@ int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) { ...@@ -1228,383 +1250,170 @@ int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) {
// build // build
n = 0; n = 0;
n = tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); for (int32_t iDelIdx = 0; iDelIdx < taosArrayGetSize(aDelIdx); iDelIdx++) {
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) { n += tPutDelIdx(pWriter->aBuf[0] + n, taosArrayGet(aDelIdx, iDelIdx));
n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size); ASSERT(n == size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write // write
n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size); code = tsdbWriteFile(pWriter->pWriteH, pWriter->fDel.size, pWriter->aBuf[0], size);
if (n < 0) { if (code) goto _err;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// update // update
pHeadFile->offset = pHeadFile->size; pWriter->fDel.offset = pWriter->fDel.size;
pHeadFile->size += size; pWriter->fDel.size += size;
_exit:
tsdbTrace("vgId:%d write block idx, offset:%" PRId64 " size:%" PRId64 " nBlockIdx:%d", TD_VID(pWriter->pTsdb->pVnode),
pHeadFile->offset, size, taosArrayGetSize(aBlockIdx));
return code; return code;
_err: _err:
tsdbError("vgId:%d, write block idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, write del idx failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) { int32_t tsdbUpdateDelFileHdr(SDelFWriter *pWriter) {
int32_t code = 0; int32_t code = 0;
SHeadFile *pHeadFile = &pWriter->fHead; char hdr[TSDB_FHDR_SIZE] = {0};
int64_t size; int64_t size = TSDB_FHDR_SIZE;
int64_t n; int64_t n;
ASSERT(mBlock->nItem > 0);
// alloc
size = sizeof(uint32_t) + tPutMapData(NULL, mBlock) + sizeof(TSCKSUM);
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err;
// build // build
n = 0; tPutDelFile(hdr, &pWriter->fDel);
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
n += tPutMapData(pWriter->aBuf[0] + n, mBlock);
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write // write
n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size); code = tsdbWriteFile(pWriter->pWriteH, 0, hdr, size);
if (n < 0) { if (code) goto _err;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
// update
pBlockIdx->offset = pHeadFile->size;
pBlockIdx->size = size;
pHeadFile->size += size;
tsdbTrace("vgId:%d, write block, file ID:%d commit ID:%d suid:%" PRId64 " uid:%" PRId64 " offset:%" PRId64
" size:%" PRId64 " nItem:%d",
TD_VID(pWriter->pTsdb->pVnode), pWriter->wSet.fid, pHeadFile->commitID, pBlockIdx->suid, pBlockIdx->uid,
pBlockIdx->offset, pBlockIdx->size, mBlock->nItem);
return code; return code;
_err: _err:
tsdbError("vgId:%d, write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, update del file hdr failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
// SDelFReader ====================================================
struct SDelFReader {
STsdb *pTsdb;
SDelFile fDel;
STsdbFD *pReadH;
uint8_t *aBuf[1];
};
int32_t tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk) { int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile, STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1]; char fname[TSDB_FILENAME_LEN];
int64_t size = 0; SDelFReader *pDelFReader;
int64_t n; int64_t n;
// check
if (taosArrayGetSize(aSstBlk) == 0) {
pSstFile->offset = pSstFile->size;
goto _exit;
}
// size
size = sizeof(uint32_t); // TSDB_FILE_DLMT
for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
size += tPutSstBlk(NULL, taosArrayGet(aSstBlk, iBlockL));
}
size += sizeof(TSCKSUM);
// alloc // alloc
code = tRealloc(&pWriter->aBuf[0], size); pDelFReader = (SDelFReader *)taosMemoryCalloc(1, sizeof(*pDelFReader));
if (code) goto _err; if (pDelFReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
// encode
n = 0;
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
n += tPutSstBlk(pWriter->aBuf[0] + n, taosArrayGet(aSstBlk, iBlockL));
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write
n = taosWriteFile(pWriter->pLastFD, pWriter->aBuf[0], size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// update // open impl
pSstFile->offset = pSstFile->size; pDelFReader->pTsdb = pTsdb;
pSstFile->size += size; pDelFReader->fDel = *pFile;
_exit: tsdbDelFileName(pTsdb, pFile, fname);
tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode), code = tsdbOpenFile(fname, TSDB_DEFAULT_PAGE_SIZE, TD_FILE_READ, &pDelFReader->pReadH);
pSstFile->offset, size); if (code) goto _err;
*ppReader = pDelFReader;
return code; return code;
_err: _err:
tsdbError("vgId:%d tsdb write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, del file reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppReader = NULL;
return code; return code;
} }
static void tsdbUpdateBlockInfo(SBlockData *pBlockData, SDataBlk *pDataBlk) { int32_t tsdbDelFReaderClose(SDelFReader **ppReader) {
for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) { int32_t code = 0;
TSDBKEY key = {.ts = pBlockData->aTSKEY[iRow], .version = pBlockData->aVersion[iRow]}; SDelFReader *pReader = *ppReader;
if (iRow == 0) {
if (tsdbKeyCmprFn(&pDataBlk->minKey, &key) > 0) {
pDataBlk->minKey = key;
}
} else {
if (pBlockData->aTSKEY[iRow] == pBlockData->aTSKEY[iRow - 1]) {
pDataBlk->hasDup = 1;
}
}
if (iRow == pBlockData->nRow - 1 && tsdbKeyCmprFn(&pDataBlk->maxKey, &key) < 0) { if (pReader) {
pDataBlk->maxKey = key; tsdbCloseFile(&pReader->pReadH);
for (int32_t iBuf = 0; iBuf < sizeof(pReader->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree(pReader->aBuf[iBuf]);
} }
taosMemoryFree(pReader);
pDataBlk->minVer = TMIN(pDataBlk->minVer, key.version);
pDataBlk->maxVer = TMAX(pDataBlk->maxVer, key.version);
} }
pDataBlk->nRow += pBlockData->nRow; *ppReader = NULL;
_exit:
return code;
} }
static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, SSmaInfo *pSmaInfo) { int32_t tsdbReadDelData(SDelFReader *pReader, SDelIdx *pDelIdx, SArray *aDelData) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pDelIdx->offset;
int64_t size = pDelIdx->size;
int64_t n;
pSmaInfo->offset = 0; taosArrayClear(aDelData);
pSmaInfo->size = 0;
// encode
for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type)) continue;
SColumnDataAgg sma;
tsdbCalcColDataSMA(pColData, &sma);
code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma));
if (code) goto _err;
pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma);
}
// write // alloc
if (pSmaInfo->size) { code = tRealloc(&pReader->aBuf[0], size);
int32_t size = pSmaInfo->size + sizeof(TSCKSUM); if (code) goto _err;
code = tRealloc(&pWriter->aBuf[0], size); // read
if (code) goto _err; code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size);
if (code) goto _err;
taosCalcChecksumAppend(0, pWriter->aBuf[0], size); // // decode
n = 0;
while (n < size) {
SDelData delData;
n += tGetDelData(pReader->aBuf[0] + n, &delData);
int64_t n = taosWriteFile(pWriter->pSmaFD, pWriter->aBuf[0], size); if (taosArrayPush(aDelData, &delData) == NULL) {
if (n < 0) { code = TSDB_CODE_OUT_OF_MEMORY;
code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
pSmaInfo->offset = pWriter->fSma.size;
pWriter->fSma.size += size;
} }
ASSERT(n == size);
return code; return code;
_err: _err:
tsdbError("vgId:%d tsdb write block sma failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, read del data failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo, int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx) {
int8_t cmprAlg, int8_t toLast) {
int32_t code = 0; int32_t code = 0;
int32_t n;
int64_t offset = pReader->fDel.offset;
int64_t size = pReader->fDel.size - offset;
ASSERT(pBlockData->nRow > 0); taosArrayClear(aDelIdx);
pBlkInfo->offset = toLast ? pWriter->fSst[pWriter->wSet.nSstF - 1].size : pWriter->fData.size;
pBlkInfo->szBlock = 0;
pBlkInfo->szKey = 0;
int32_t aBufN[4] = {0}; // alloc
code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN); code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// write ================= // read
TdFilePtr pFD = toLast ? pWriter->pLastFD : pWriter->pDataFD; code = tsdbReadFile(pReader->pReadH, offset, pReader->aBuf[0], size);
if (code) goto _err;
pBlkInfo->szKey = aBufN[3] + aBufN[2];
pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
int64_t n = taosWriteFile(pFD, pWriter->aBuf[3], aBufN[3]);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pFD, pWriter->aBuf[2], aBufN[2]); // decode
if (n < 0) { n = 0;
code = TAOS_SYSTEM_ERROR(errno); while (n < size) {
goto _err; SDelIdx delIdx;
}
if (aBufN[1]) { n += tGetDelIdx(pReader->aBuf[0] + n, &delIdx);
n = taosWriteFile(pFD, pWriter->aBuf[1], aBufN[1]);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
if (aBufN[0]) { if (taosArrayPush(aDelIdx, &delIdx) == NULL) {
n = taosWriteFile(pFD, pWriter->aBuf[0], aBufN[0]); code = TSDB_CODE_OUT_OF_MEMORY;
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
} }
// update info ASSERT(n == size);
if (toLast) {
pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock;
} else {
pWriter->fData.size += pBlkInfo->szBlock;
}
// ================= SMA ====================
if (pSmaInfo) {
code = tsdbWriteBlockSma(pWriter, pBlockData, pSmaInfo);
if (code) goto _err;
}
_exit:
tsdbTrace("vgId:%d tsdb write block data, suid:%" PRId64 " uid:%" PRId64 " nRow:%d, offset:%" PRId64 " size:%d",
TD_VID(pWriter->pTsdb->pVnode), pBlockData->suid, pBlockData->uid, pBlockData->nRow, pBlkInfo->offset,
pBlkInfo->szBlock);
return code;
_err:
tsdbError("vgId:%d tsdb write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code;
}
int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
int32_t code = 0;
int64_t n;
int64_t size;
TdFilePtr pOutFD = NULL; // TODO
TdFilePtr PInFD = NULL; // TODO
char fNameFrom[TSDB_FILENAME_LEN];
char fNameTo[TSDB_FILENAME_LEN];
// head
tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom);
tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// data
tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom);
tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sst
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) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sma
tsdbSmaFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pSmaF, fNameFrom);
tsdbSmaFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pSmaF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pOutFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
return code; return code;
_err: _err:
tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d, read del idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbUtil.c
浏览文件 @ 0c27b07c
...@@ -1548,7 +1548,7 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut, ...@@ -1548,7 +1548,7 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
if (code) goto _exit; if (code) goto _exit;
blockCol.offset = aBufN[0]; blockCol.offset = aBufN[0];
aBufN[0] = aBufN[0] + blockCol.szBitmap + blockCol.szOffset + blockCol.szValue + sizeof(TSCKSUM); aBufN[0] = aBufN[0] + blockCol.szBitmap + blockCol.szOffset + blockCol.szValue;
} }
code = tRealloc(&aBuf[1], hdr.szBlkCol + tPutBlockCol(NULL, &blockCol)); code = tRealloc(&aBuf[1], hdr.szBlkCol + tPutBlockCol(NULL, &blockCol));
...@@ -1556,15 +1556,8 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut, ...@@ -1556,15 +1556,8 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
hdr.szBlkCol += tPutBlockCol(aBuf[1] + hdr.szBlkCol, &blockCol); hdr.szBlkCol += tPutBlockCol(aBuf[1] + hdr.szBlkCol, &blockCol);
} }
aBufN[1] = 0; // SBlockCol
if (hdr.szBlkCol > 0) { aBufN[1] = hdr.szBlkCol;
aBufN[1] = hdr.szBlkCol + sizeof(TSCKSUM);
code = tRealloc(&aBuf[1], aBufN[1]);
if (code) goto _exit;
taosCalcChecksumAppend(0, aBuf[1], aBufN[1]);
}
// uid + version + tskey // uid + version + tskey
aBufN[2] = 0; aBufN[2] = 0;
...@@ -1585,16 +1578,11 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut, ...@@ -1585,16 +1578,11 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
if (code) goto _exit; if (code) goto _exit;
aBufN[2] += hdr.szKey; aBufN[2] += hdr.szKey;
aBufN[2] += sizeof(TSCKSUM);
code = tRealloc(&aBuf[2], aBufN[2]);
if (code) goto _exit;
// hdr // hdr
aBufN[3] = tPutDiskDataHdr(NULL, &hdr); aBufN[3] = tPutDiskDataHdr(NULL, &hdr);
code = tRealloc(&aBuf[3], aBufN[3]); code = tRealloc(&aBuf[3], aBufN[3]);
if (code) goto _exit; if (code) goto _exit;
tPutDiskDataHdr(aBuf[3], &hdr); tPutDiskDataHdr(aBuf[3], &hdr);
taosCalcChecksumAppend(taosCalcChecksum(0, aBuf[3], aBufN[3]), aBuf[2], aBufN[2]);
// aggragate // aggragate
if (ppOut) { if (ppOut) {
...@@ -1626,10 +1614,6 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin ...@@ -1626,10 +1614,6 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
// SDiskDataHdr // SDiskDataHdr
n += tGetDiskDataHdr(pIn + n, &hdr); n += tGetDiskDataHdr(pIn + n, &hdr);
if (!taosCheckChecksumWhole(pIn, n + hdr.szUid + hdr.szVer + hdr.szKey + sizeof(TSCKSUM))) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
ASSERT(hdr.delimiter == TSDB_FILE_DLMT); ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
pBlockData->suid = hdr.suid; pBlockData->suid = hdr.suid;
...@@ -1657,7 +1641,7 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin ...@@ -1657,7 +1641,7 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
code = tsdbDecmprData(pIn + n, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY, code = tsdbDecmprData(pIn + n, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
sizeof(TSKEY) * hdr.nRow, &aBuf[0]); sizeof(TSKEY) * hdr.nRow, &aBuf[0]);
if (code) goto _exit; if (code) goto _exit;
n = n + hdr.szKey + sizeof(TSCKSUM); n += hdr.szKey;
// loop to decode each column data // loop to decode each column data
if (hdr.szBlkCol == 0) goto _exit; if (hdr.szBlkCol == 0) goto _exit;
...@@ -1679,8 +1663,8 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin ...@@ -1679,8 +1663,8 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
if (code) goto _exit; if (code) goto _exit;
} }
} else { } else {
code = tsdbDecmprColData(pIn + n + hdr.szBlkCol + sizeof(TSCKSUM) + blockCol.offset, &blockCol, hdr.cmprAlg, code = tsdbDecmprColData(pIn + n + hdr.szBlkCol + blockCol.offset, &blockCol, hdr.cmprAlg, hdr.nRow, pColData,
hdr.nRow, pColData, &aBuf[0]); &aBuf[0]);
if (code) goto _exit; if (code) goto _exit;
} }
} }
...@@ -2062,12 +2046,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol ...@@ -2062,12 +2046,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
} }
size += pBlockCol->szValue; size += pBlockCol->szValue;
// checksum
size += sizeof(TSCKSUM);
code = tRealloc(ppOut, nOut + size);
if (code) goto _exit;
taosCalcChecksumAppend(0, *ppOut + nOut, size);
_exit: _exit:
return code; return code;
} }
...@@ -2076,12 +2054,6 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in ...@@ -2076,12 +2054,6 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in
uint8_t **ppBuf) { uint8_t **ppBuf) {
int32_t code = 0; int32_t code = 0;
int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
if (!taosCheckChecksumWhole(pIn, size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
ASSERT(pColData->cid == pBlockCol->cid); ASSERT(pColData->cid == pBlockCol->cid);
ASSERT(pColData->type == pBlockCol->type); ASSERT(pColData->type == pBlockCol->type);
pColData->smaOn = pBlockCol->smaOn; pColData->smaOn = pBlockCol->smaOn;
...@@ -2153,37 +2125,3 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in ...@@ -2153,37 +2125,3 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in
_exit: _exit:
return code; return code;
} }
int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck) {
int32_t code = 0;
// alloc
code = tRealloc(ppOut, size);
if (code) goto _exit;
// seek
int64_t n = taosLSeekFile(pFD, offset, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
// read
n = taosReadFile(pFD, *ppOut, size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
// check
if (toCheck && !taosCheckChecksumWhole(*ppOut, size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
_exit:
return code;
}
utils/test/c/sml_test.c
浏览文件 @ 0c27b07c
...@@ -92,7 +92,7 @@ int smlProcess_telnet_Test() { ...@@ -92,7 +92,7 @@ int smlProcess_telnet_Test() {
int smlProcess_json1_Test() { int smlProcess_json1_Test() {
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db schemaless 1"); TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db");
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(taos, "use sml_db"); pRes = taos_query(taos, "use sml_db");
...@@ -112,7 +112,7 @@ int smlProcess_json1_Test() { ...@@ -112,7 +112,7 @@ int smlProcess_json1_Test() {
" }," " },"
" {" " {"
" \"metric\": \"sys.cpu.nice\"," " \"metric\": \"sys.cpu.nice\","
" \"timestamp\": 1346846400," " \"timestamp\": 1662344042,"
" \"value\": 9," " \"value\": 9,"
" \"tags\": {" " \"tags\": {"
" \"host\": \"web02\"," " \"host\": \"web02\","
...@@ -141,7 +141,7 @@ int smlProcess_json2_Test() { ...@@ -141,7 +141,7 @@ int smlProcess_json2_Test() {
"{" "{"
" \"metric\": \"meter_current0\"," " \"metric\": \"meter_current0\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846400," " \"value\" : 1662344042,"
" \"type\" : \"s\"" " \"type\" : \"s\""
" }," " },"
" \"value\": {" " \"value\": {"
...@@ -181,7 +181,7 @@ int smlProcess_json3_Test() { ...@@ -181,7 +181,7 @@ int smlProcess_json3_Test() {
"{" "{"
" \"metric\": \"meter_current1\"," " \"metric\": \"meter_current1\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846400," " \"value\" : 1662344042,"
" \"type\" : \"s\"" " \"type\" : \"s\""
" }," " },"
" \"value\": {" " \"value\": {"
...@@ -249,7 +249,7 @@ int smlProcess_json4_Test() { ...@@ -249,7 +249,7 @@ int smlProcess_json4_Test() {
"{" "{"
" \"metric\": \"meter_current2\"," " \"metric\": \"meter_current2\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846500000," " \"value\" : 1662344042000,"
" \"type\" : \"ms\"" " \"type\" : \"ms\""
" }," " },"
" \"value\": \"ni\"," " \"value\": \"ni\","
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或