Skip to content

TDengine
TDengine

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

通知 1192
Star 22018
Fork 4786
TDengine
TDengine
提交 8329284a 编写于 作者: H Hongze Cheng
浏览文件
操作

refact for further dev

上级 7fe743d0
隐藏空白更改
内联 并排
Showing with 804 addition and 829 deletion
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 8329284a
...@@ -586,6 +586,12 @@ struct SDelFWriter { ...@@ -586,6 +586,12 @@ struct SDelFWriter {
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;
...@@ -603,12 +609,6 @@ struct SDataFWriter { ...@@ -603,12 +609,6 @@ struct SDataFWriter {
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;
......
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
浏览文件 @ 8329284a
...@@ -23,7 +23,7 @@ typedef struct { ...@@ -23,7 +23,7 @@ typedef struct {
int64_t pgno; int64_t pgno;
} STsdbFD; } STsdbFD;
int32_t tsdbOpenFile(const char *path, int32_t opt, STsdbFD *pFD) { static int32_t tsdbOpenFile(const char *path, int32_t opt, STsdbFD *pFD) {
int32_t code = 0; int32_t code = 0;
pFD->pFD = taosOpenFile(path, opt); pFD->pFD = taosOpenFile(path, opt);
...@@ -45,12 +45,12 @@ _exit: ...@@ -45,12 +45,12 @@ _exit:
return code; return code;
} }
void tsdbCloseFile(STsdbFD *pFD) { static void tsdbCloseFile(STsdbFD *pFD) {
taosMemoryFree(pFD->pBuf); taosMemoryFree(pFD->pBuf);
taosCloseFile(&pFD->pFD); taosCloseFile(&pFD->pFD);
} }
int32_t tsdbSyncFile(STsdbFD *pFD) { static int32_t tsdbSyncFile(STsdbFD *pFD) {
int32_t code = 0; int32_t code = 0;
if (taosFsyncFile(pFD->pFD) < 0) { if (taosFsyncFile(pFD->pFD) < 0) {
...@@ -62,7 +62,7 @@ _exit: ...@@ -62,7 +62,7 @@ _exit:
return code; return code;
} }
int32_t tsdbWriteFile(STsdbFD *pFD, uint8_t *pBuf, int32_t nBuf, int64_t *offset) { static int32_t tsdbWriteFile(STsdbFD *pFD, uint8_t *pBuf, int32_t nBuf, int64_t *offset) {
int32_t code = 0; int32_t code = 0;
int32_t n = 0; int32_t n = 0;
...@@ -120,7 +120,7 @@ _exit: ...@@ -120,7 +120,7 @@ _exit:
return code; return code;
} }
int64_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t count) { static int64_t tsdbReadFile(STsdbFD *pFD, int64_t offset, uint8_t *pBuf, int64_t count) {
int32_t code = 0; int32_t code = 0;
int64_t pgno = offset / pFD->szPage; int64_t pgno = offset / pFD->szPage;
...@@ -147,1219 +147,1195 @@ _exit: ...@@ -147,1219 +147,1195 @@ _exit:
return code; return code;
} }
// SDataFReader ==================================================== // SDataFWriter ====================================================
int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) { int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) {
int32_t code = 0; int32_t code = 0;
SDataFReader *pReader; int32_t flag;
int64_t n;
SDataFWriter *pWriter = NULL;
char fname[TSDB_FILENAME_LEN]; char fname[TSDB_FILENAME_LEN];
char hdr[TSDB_FHDR_SIZE] = {0};
// alloc // alloc
pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader)); pWriter = taosMemoryCalloc(1, sizeof(*pWriter));
if (pReader == NULL) { if (pWriter == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
pReader->pTsdb = pTsdb; pWriter->pTsdb = pTsdb;
pReader->pSet = pSet; 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];
}
// open impl
// head // head
tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname); flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
pReader->pHeadFD = taosOpenFile(fname, TD_FILE_READ); tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname);
if (pReader->pHeadFD == NULL) { pWriter->pHeadFD = taosOpenFile(fname, flag);
if (pWriter->pHeadFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
ASSERT(n == TSDB_FHDR_SIZE);
pWriter->fHead.size += TSDB_FHDR_SIZE;
// data // data
tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname); if (pWriter->fData.size == 0) {
pReader->pDataFD = taosOpenFile(fname, TD_FILE_READ); flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
if (pReader->pDataFD == NULL) { } 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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; 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 // sma
tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname); if (pWriter->fSma.size == 0) {
pReader->pSmaFD = taosOpenFile(fname, TD_FILE_READ); flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
if (pReader->pSmaFD == NULL) { } 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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; 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;
}
// sst pWriter->fSma.size += TSDB_FHDR_SIZE;
for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) { } else {
tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname); n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_END);
pReader->aLastFD[iSst] = taosOpenFile(fname, TD_FILE_READ); if (n < 0) {
if (pReader->aLastFD[iSst] == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
ASSERT(n == pWriter->fSma.size);
} }
*ppReader = pReader; // 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; 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 data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppReader = NULL; *ppWriter = NULL;
return code; return code;
} }
int32_t tsdbDataFReaderClose(SDataFReader **ppReader) { int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
int32_t code = 0; int32_t code = 0;
if (*ppReader == NULL) goto _exit; STsdb *pTsdb = NULL;
// head if (*ppWriter == NULL) goto _exit;
if (taosCloseFile(&(*ppReader)->pHeadFD) < 0) {
pTsdb = (*ppWriter)->pTsdb;
if (sync) {
if (taosFsyncFile((*ppWriter)->pHeadFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// data if (taosCloseFile(&(*ppWriter)->pDataFD) < 0) {
if (taosCloseFile(&(*ppReader)->pDataFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// sma if (taosCloseFile(&(*ppWriter)->pSmaFD) < 0) {
if (taosCloseFile(&(*ppReader)->pSmaFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// sst if (taosCloseFile(&(*ppWriter)->pLastFD) < 0) {
for (int32_t iSst = 0; iSst < (*ppReader)->pSet->nSstF; iSst++) { code = TAOS_SYSTEM_ERROR(errno);
if (taosCloseFile(&(*ppReader)->aLastFD[iSst]) < 0) { goto _err;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
} }
for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) { for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree((*ppReader)->aBuf[iBuf]); tFree((*ppWriter)->aBuf[iBuf]);
} }
taosMemoryFree(*ppReader); taosMemoryFree(*ppWriter);
_exit: _exit:
*ppReader = NULL; *ppWriter = 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, data file writer close failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) { int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pReader->pSet->pHeadF->offset; int64_t n;
int64_t size = pReader->pSet->pHeadF->size - offset; char hdr[TSDB_FHDR_SIZE];
int64_t n;
uint32_t delimiter;
taosArrayClear(aBlockIdx);
if (size == 0) {
goto _exit;
}
// alloc // head ==============
code = tRealloc(&pReader->aBuf[0], size); memset(hdr, 0, TSDB_FHDR_SIZE);
if (code) goto _err; tPutHeadFile(hdr, &pWriter->fHead);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
// seek n = taosLSeekFile(pWriter->pHeadFD, 0, SEEK_SET);
if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// read n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_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;
// 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; goto _err;
} }
// check n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE);
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { if (n < 0) {
code = TSDB_CODE_FILE_CORRUPTED; code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// decode // sma ==============
n = 0; memset(hdr, 0, TSDB_FHDR_SIZE);
n = tGetU32(pReader->aBuf[0] + n, &delimiter); tPutSmaFile(hdr, &pWriter->fSma);
ASSERT(delimiter == TSDB_FILE_DLMT); taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
while (n < size - sizeof(TSCKSUM)) { n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_SET);
SBlockIdx blockIdx; if (n < 0) {
n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx); code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) { n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE);
code = TSDB_CODE_OUT_OF_MEMORY; if (n < 0) {
goto _err; code = TAOS_SYSTEM_ERROR(errno);
} goto _err;
} }
ASSERT(n + sizeof(TSCKSUM) == size); // sst ==============
memset(hdr, 0, TSDB_FHDR_SIZE);
tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]);
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);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
_exit:
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, update DFileSet header 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 tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pReader->pSet->aSstF[iSst]->offset; SHeadFile *pHeadFile = &pWriter->fHead;
int64_t size = pReader->pSet->aSstF[iSst]->size - offset; int64_t size = 0;
int64_t n; int64_t n;
uint32_t delimiter;
taosArrayClear(aSstBlk); // check
if (size == 0) { if (taosArrayGetSize(aBlockIdx) == 0) {
pHeadFile->offset = pHeadFile->size;
goto _exit; goto _exit;
} }
// prepare
size = sizeof(uint32_t);
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
}
size += sizeof(TSCKSUM);
// alloc // alloc
code = tRealloc(&pReader->aBuf[0], size); code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// seek // build
if (taosLSeekFile(pReader->aLastFD[iSst], offset, SEEK_SET) < 0) { n = 0;
code = TAOS_SYSTEM_ERROR(errno); n = tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
goto _err; for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
// read ASSERT(n + sizeof(TSCKSUM) == size);
n = taosReadFile(pReader->aLastFD[iSst], pReader->aBuf[0], size);
// write
n = taosWriteFile(pWriter->pHeadFD, pWriter->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;
} }
// check // update
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { pHeadFile->offset = pHeadFile->size;
code = TSDB_CODE_FILE_CORRUPTED; pHeadFile->size += size;
goto _err;
}
// decode
n = 0;
n = tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) {
SSstBlk blockl;
n += tGetSstBlk(pReader->aBuf[0] + n, &blockl);
if (taosArrayPush(aSstBlk, &blockl) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
ASSERT(n + sizeof(TSCKSUM) == size);
_exit: _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 blockl 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 tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) { int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) {
int32_t code = 0; int32_t code = 0;
int64_t offset = pBlockIdx->offset; SHeadFile *pHeadFile = &pWriter->fHead;
int64_t size = pBlockIdx->size; int64_t size;
int64_t n; int64_t n;
int64_t tn;
ASSERT(mBlock->nItem > 0);
// alloc // alloc
code = tRealloc(&pReader->aBuf[0], size); size = sizeof(uint32_t) + tPutMapData(NULL, mBlock) + sizeof(TSCKSUM);
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// seek // build
if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) { n = 0;
code = TAOS_SYSTEM_ERROR(errno); n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
goto _err; n += tPutMapData(pWriter->aBuf[0] + n, mBlock);
} taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
// read ASSERT(n + sizeof(TSCKSUM) == size);
n = taosReadFile(pReader->pHeadFD, pReader->aBuf[0], size);
// write
n = taosWriteFile(pWriter->pHeadFD, pWriter->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;
} }
// check // update
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) { pBlockIdx->offset = pHeadFile->size;
code = TSDB_CODE_FILE_CORRUPTED; pBlockIdx->size = size;
goto _err; pHeadFile->size += size;
}
// decode
n = 0;
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;
}
n += tn;
ASSERT(n + sizeof(TSCKSUM) == 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 block 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;
} }
int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aColumnDataAgg) { int32_t tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk) {
int32_t code = 0; int32_t code = 0;
SSmaInfo *pSmaInfo = &pDataBlk->smaInfo; SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1];
int64_t size;
int64_t n;
ASSERT(pSmaInfo->size > 0); // check
if (taosArrayGetSize(aSstBlk) == 0) {
pSstFile->offset = pSstFile->size;
goto _exit;
}
taosArrayClear(aColumnDataAgg); // 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
int32_t size = pSmaInfo->size + sizeof(TSCKSUM); code = tRealloc(&pWriter->aBuf[0], size);
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// seek // encode
int64_t n = taosLSeekFile(pReader->pSmaFD, pSmaInfo->offset, SEEK_SET); n = 0;
if (n < 0) { n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT);
code = TAOS_SYSTEM_ERROR(errno); for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) {
goto _err; n += tPutSstBlk(pWriter->aBuf[0] + n, taosArrayGet(aSstBlk, iBlockL));
} else if (n < pSmaInfo->offset) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
// read ASSERT(n + sizeof(TSCKSUM) == size);
n = taosReadFile(pReader->pSmaFD, pReader->aBuf[0], size);
// write
n = taosWriteFile(pWriter->pLastFD, pWriter->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;
}
// check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
// decode // update
n = 0; pSstFile->offset = pSstFile->size;
while (n < pSmaInfo->size) { pSstFile->size += size;
SColumnDataAgg sma;
n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
_exit:
tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
pSstFile->offset, size);
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 write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, int8_t fromLast, static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, SSmaInfo *pSmaInfo) {
SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
tBlockDataClear(pBlockData); pSmaInfo->offset = 0;
pSmaInfo->size = 0;
TdFilePtr pFD = fromLast ? pReader->aLastFD[0] : pReader->pDataFD; // (todo) // encode
for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
// uid + version + tskey if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type)) continue;
code = tsdbReadAndCheck(pFD, pBlkInfo->offset, &pReader->aBuf[0], pBlkInfo->szKey, 1);
if (code) goto _err;
SDiskDataHdr hdr;
uint8_t *p = pReader->aBuf[0] + tGetDiskDataHdr(pReader->aBuf[0], &hdr);
ASSERT(hdr.delimiter == TSDB_FILE_DLMT); SColumnDataAgg sma;
ASSERT(pBlockData->suid == hdr.suid); tsdbCalcColDataSMA(pColData, &sma);
ASSERT(pBlockData->uid == hdr.uid);
pBlockData->nRow = hdr.nRow; code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma));
if (code) goto _err;
pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma);
}
// uid // write
if (hdr.uid == 0) { if (pSmaInfo->size) {
ASSERT(hdr.szUid); int32_t size = pSmaInfo->size + sizeof(TSCKSUM);
code = tsdbDecmprData(p, hdr.szUid, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aUid,
sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err;
} else {
ASSERT(!hdr.szUid);
}
p += hdr.szUid;
// version
code = tsdbDecmprData(p, hdr.szVer, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aVersion,
sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err;
p += hdr.szVer;
// TSKEY
code = tsdbDecmprData(p, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
sizeof(TSKEY) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err;
p += hdr.szKey;
ASSERT(p - pReader->aBuf[0] == pBlkInfo->szKey - sizeof(TSCKSUM));
// read and decode columns
if (taosArrayGetSize(pBlockData->aIdx) == 0) goto _exit;
if (hdr.szBlkCol > 0) { code = tRealloc(&pWriter->aBuf[0], size);
int64_t offset = pBlkInfo->offset + pBlkInfo->szKey;
code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[0], hdr.szBlkCol + sizeof(TSCKSUM), 1);
if (code) goto _err; if (code) goto _err;
}
SBlockCol blockCol = {.cid = 0};
SBlockCol *pBlockCol = &blockCol;
int32_t n = 0;
for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) { taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
while (pBlockCol && pBlockCol->cid < pColData->cid) { int64_t n = taosWriteFile(pWriter->pSmaFD, pWriter->aBuf[0], size);
if (n < hdr.szBlkCol) { if (n < 0) {
n += tGetBlockCol(pReader->aBuf[0] + n, pBlockCol); code = TAOS_SYSTEM_ERROR(errno);
} else { goto _err;
ASSERT(n == hdr.szBlkCol);
pBlockCol = NULL;
}
} }
if (pBlockCol == NULL || pBlockCol->cid > pColData->cid) { pSmaInfo->offset = pWriter->fSma.size;
// add a lot of NONE pWriter->fSma.size += size;
for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
if (code) goto _err;
}
} else {
ASSERT(pBlockCol->type == pColData->type);
ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);
if (pBlockCol->flag == HAS_NULL) {
// add a lot of NULL
for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
if (code) goto _err;
}
} else {
// decode from binary
int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + sizeof(TSCKSUM) + pBlockCol->offset;
int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[1], size, 0);
if (code) goto _err;
code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]);
if (code) goto _err;
}
}
} }
_exit:
return code; return code;
_err: _err:
tsdbError("vgId:%d tsdb read block data impl failed since %s", TD_VID(pReader->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 tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) { int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
int8_t cmprAlg, int8_t toLast) {
int32_t code = 0; int32_t code = 0;
code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], 0, pBlockData); ASSERT(pBlockData->nRow > 0);
if (code) goto _err;
if (pDataBlk->nSubBlock > 1) {
SBlockData bData1;
SBlockData bData2;
// create pBlkInfo->offset = toLast ? pWriter->fSst[pWriter->wSet.nSstF - 1].size : pWriter->fData.size;
code = tBlockDataCreate(&bData1); pBlkInfo->szBlock = 0;
if (code) goto _err; pBlkInfo->szKey = 0;
code = tBlockDataCreate(&bData2);
if (code) goto _err;
// init int32_t aBufN[4] = {0};
tBlockDataInitEx(&bData1, pBlockData); code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN);
tBlockDataInitEx(&bData2, pBlockData); if (code) goto _err;
for (int32_t iSubBlock = 1; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) { // write =================
code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[iSubBlock], 0, &bData1); TdFilePtr pFD = toLast ? pWriter->pLastFD : pWriter->pDataFD;
if (code) {
tBlockDataDestroy(&bData1, 1);
tBlockDataDestroy(&bData2, 1);
goto _err;
}
code = tBlockDataCopy(pBlockData, &bData2); pBlkInfo->szKey = aBufN[3] + aBufN[2];
if (code) { pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3];
tBlockDataDestroy(&bData1, 1);
tBlockDataDestroy(&bData2, 1);
goto _err;
}
code = tBlockDataMerge(&bData1, &bData2, pBlockData); int64_t n = taosWriteFile(pFD, pWriter->aBuf[3], aBufN[3]);
if (code) { if (n < 0) {
tBlockDataDestroy(&bData1, 1); code = TAOS_SYSTEM_ERROR(errno);
tBlockDataDestroy(&bData2, 1); goto _err;
goto _err; }
}
}
tBlockDataDestroy(&bData1, 1); n = taosWriteFile(pFD, pWriter->aBuf[2], aBufN[2]);
tBlockDataDestroy(&bData2, 1); if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
return code; if (aBufN[1]) {
n = taosWriteFile(pFD, pWriter->aBuf[1], aBufN[1]);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
_err: if (aBufN[0]) {
tsdbError("vgId:%d tsdb read data block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); n = taosWriteFile(pFD, pWriter->aBuf[0], aBufN[0]);
return code; if (n < 0) {
} code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) { // update info
int32_t code = 0; if (toLast) {
pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock;
} else {
pWriter->fData.size += pBlkInfo->szBlock;
}
code = tsdbReadBlockDataImpl(pReader, &pSstBlk->bInfo, 1, pBlockData); // ================= SMA ====================
if (code) goto _err; 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; return code;
_err: _err:
tsdbError("vgId:%d tsdb read last block 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 tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) { int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) {
int32_t code = 0; int32_t code = 0;
int64_t n;
// read int64_t size;
code = tsdbReadAndCheck(pReader->aLastFD[iSst], pSstBlk->bInfo.offset, &pReader->aBuf[0], pSstBlk->bInfo.szBlock, 0); TdFilePtr pOutFD = NULL; // TODO
if (code) goto _exit; TdFilePtr PInFD = NULL; // TODO
char fNameFrom[TSDB_FILENAME_LEN];
// decmpr char fNameTo[TSDB_FILENAME_LEN];
code = tDecmprBlockData(pReader->aBuf[0], pSstBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]);
if (code) goto _exit;
_exit:
return code;
}
// SDataFWriter ==================================================== // head
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet) { tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom);
int32_t code = 0; tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
int32_t flag;
int64_t n;
SDataFWriter *pWriter = NULL;
char fname[TSDB_FILENAME_LEN];
char hdr[TSDB_FHDR_SIZE] = {0};
// alloc pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
pWriter = taosMemoryCalloc(1, sizeof(*pWriter)); if (pOutFD == NULL) {
if (pWriter == NULL) { code = TAOS_SYSTEM_ERROR(errno);
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; 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];
}
// head PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC; if (PInFD == NULL) {
tsdbHeadFileName(pTsdb, pWriter->wSet.diskId, pWriter->wSet.fid, &pWriter->fHead, fname);
pWriter->pHeadFD = taosOpenFile(fname, flag);
if (pWriter->pHeadFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE); n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
taosCloseFile(&pOutFD);
ASSERT(n == TSDB_FHDR_SIZE); taosCloseFile(&PInFD);
pWriter->fHead.size += TSDB_FHDR_SIZE;
// data // data
if (pWriter->fData.size == 0) { tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom);
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC; tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo);
} 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 pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (pWriter->fSma.size == 0) { if (pOutFD == NULL) {
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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; 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); PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
} if (PInFD == NULL) {
// 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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE);
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size);
if (n < 0) { if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
pWriter->fSst[pWriter->wSet.nSstF - 1].size += TSDB_FHDR_SIZE; taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
*ppWriter = pWriter;
return code;
_err:
tsdbError("vgId:%d, tsdb data file writer open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
*ppWriter = NULL;
return code;
}
int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync) {
int32_t code = 0;
STsdb *pTsdb = NULL;
if (*ppWriter == NULL) goto _exit;
pTsdb = (*ppWriter)->pTsdb;
if (sync) {
if (taosFsyncFile((*ppWriter)->pHeadFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile((*ppWriter)->pDataFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile((*ppWriter)->pSmaFD) < 0) { // sst
code = TAOS_SYSTEM_ERROR(errno); tsdbSstFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->aSstF[0], fNameFrom);
goto _err; tsdbSstFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->aSstF[0], fNameTo);
}
if (taosFsyncFile((*ppWriter)->pLastFD) < 0) { pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
code = TAOS_SYSTEM_ERROR(errno); if (pOutFD == NULL) {
goto _err; code = TAOS_SYSTEM_ERROR(errno);
} goto _err;
} }
if (taosCloseFile(&(*ppWriter)->pHeadFD) < 0) { PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (taosCloseFile(&(*ppWriter)->pDataFD) < 0) { n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
if (taosCloseFile(&(*ppWriter)->pSmaFD) < 0) { // 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); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (taosCloseFile(&(*ppWriter)->pLastFD) < 0) { PInFD = taosOpenFile(fNameFrom, TD_FILE_READ);
if (PInFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
for (int32_t iBuf = 0; iBuf < sizeof((*ppWriter)->aBuf) / sizeof(uint8_t *); iBuf++) { n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size);
tFree((*ppWriter)->aBuf[iBuf]); if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
taosMemoryFree(*ppWriter); taosCloseFile(&pOutFD);
_exit: taosCloseFile(&PInFD);
*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, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter) { // SDataFReader ====================================================
int32_t code = 0; int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pSet) {
int64_t n; int32_t code = 0;
char hdr[TSDB_FHDR_SIZE]; SDataFReader *pReader;
char fname[TSDB_FILENAME_LEN];
// head ==============
memset(hdr, 0, TSDB_FHDR_SIZE);
tPutHeadFile(hdr, &pWriter->fHead);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pHeadFD, 0, SEEK_SET); // alloc
if (n < 0) { pReader = (SDataFReader *)taosMemoryCalloc(1, sizeof(*pReader));
code = TAOS_SYSTEM_ERROR(errno); if (pReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
pReader->pTsdb = pTsdb;
pReader->pSet = pSet;
n = taosWriteFile(pWriter->pHeadFD, hdr, TSDB_FHDR_SIZE); // open impl
if (n < 0) { // head
tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
pReader->pHeadFD = taosOpenFile(fname, TD_FILE_READ);
if (pReader->pHeadFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// data ============== // data
memset(hdr, 0, TSDB_FHDR_SIZE); tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
tPutDataFile(hdr, &pWriter->fData); pReader->pDataFD = taosOpenFile(fname, TD_FILE_READ);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE); if (pReader->pDataFD == NULL) {
n = taosLSeekFile(pWriter->pDataFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosWriteFile(pWriter->pDataFD, hdr, TSDB_FHDR_SIZE); // sma
if (n < 0) { tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
pReader->pSmaFD = taosOpenFile(fname, TD_FILE_READ);
if (pReader->pSmaFD == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// sma ============== // sst
memset(hdr, 0, TSDB_FHDR_SIZE); for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
tPutSmaFile(hdr, &pWriter->fSma); tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE); pReader->aLastFD[iSst] = taosOpenFile(fname, TD_FILE_READ);
if (pReader->aLastFD[iSst] == NULL) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
n = taosLSeekFile(pWriter->pSmaFD, 0, SEEK_SET); *ppReader = pReader;
if (n < 0) { 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) goto _exit;
// head
if (taosCloseFile(&(*ppReader)->pHeadFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosWriteFile(pWriter->pSmaFD, hdr, TSDB_FHDR_SIZE); // data
if (n < 0) { if (taosCloseFile(&(*ppReader)->pDataFD) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
// sst ============== // sma
memset(hdr, 0, TSDB_FHDR_SIZE); if (taosCloseFile(&(*ppReader)->pSmaFD) < 0) {
tPutSstFile(hdr, &pWriter->fSst[pWriter->wSet.nSstF - 1]);
taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
n = taosLSeekFile(pWriter->pLastFD, 0, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
n = taosWriteFile(pWriter->pLastFD, hdr, TSDB_FHDR_SIZE); // sst
if (n < 0) { for (int32_t iSst = 0; iSst < (*ppReader)->pSet->nSstF; iSst++) {
code = TAOS_SYSTEM_ERROR(errno); if (taosCloseFile(&(*ppReader)->aLastFD[iSst]) < 0) {
goto _err; code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
for (int32_t iBuf = 0; iBuf < sizeof((*ppReader)->aBuf) / sizeof(uint8_t *); iBuf++) {
tFree((*ppReader)->aBuf[iBuf]);
} }
taosMemoryFree(*ppReader);
_exit:
*ppReader = NULL;
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, data file reader close failed since %s", TD_VID((*ppReader)->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx) { int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx) {
int32_t code = 0; int32_t code = 0;
SHeadFile *pHeadFile = &pWriter->fHead; int64_t offset = pReader->pSet->pHeadF->offset;
int64_t size = 0; int64_t size = pReader->pSet->pHeadF->size - offset;
int64_t n; int64_t n;
uint32_t delimiter;
// check taosArrayClear(aBlockIdx);
if (taosArrayGetSize(aBlockIdx) == 0) { if (size == 0) {
pHeadFile->offset = pHeadFile->size;
goto _exit; goto _exit;
} }
// prepare
size = sizeof(uint32_t);
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) {
size += tPutBlockIdx(NULL, taosArrayGet(aBlockIdx, iBlockIdx));
}
size += sizeof(TSCKSUM);
// alloc // alloc
code = tRealloc(&pWriter->aBuf[0], size); code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// build // seek
n = 0; if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) {
n = tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); code = TAOS_SYSTEM_ERROR(errno);
for (int32_t iBlockIdx = 0; iBlockIdx < taosArrayGetSize(aBlockIdx); iBlockIdx++) { goto _err;
n += tPutBlockIdx(pWriter->aBuf[0] + n, taosArrayGet(aBlockIdx, iBlockIdx));
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write // read
n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], 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;
}
// check
if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
// decode
n = 0;
n = tGetU32(pReader->aBuf[0] + n, &delimiter);
ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) {
SBlockIdx blockIdx;
n += tGetBlockIdx(pReader->aBuf[0] + n, &blockIdx);
if (taosArrayPush(aBlockIdx, &blockIdx) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
} }
// update ASSERT(n + sizeof(TSCKSUM) == size);
pHeadFile->offset = pHeadFile->size;
pHeadFile->size += size;
_exit: _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, read block idx failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, SBlockIdx *pBlockIdx) { int32_t tsdbReadSstBlk(SDataFReader *pReader, int32_t iSst, SArray *aSstBlk) {
int32_t code = 0; int32_t code = 0;
SHeadFile *pHeadFile = &pWriter->fHead; int64_t offset = pReader->pSet->aSstF[iSst]->offset;
int64_t size; int64_t size = pReader->pSet->aSstF[iSst]->size - offset;
int64_t n; int64_t n;
uint32_t delimiter;
ASSERT(mBlock->nItem > 0); taosArrayClear(aSstBlk);
if (size == 0) {
goto _exit;
}
// alloc // alloc
size = sizeof(uint32_t) + tPutMapData(NULL, mBlock) + sizeof(TSCKSUM); code = tRealloc(&pReader->aBuf[0], size);
code = tRealloc(&pWriter->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// build // seek
n = 0; if (taosLSeekFile(pReader->aLastFD[iSst], offset, SEEK_SET) < 0) {
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); code = TAOS_SYSTEM_ERROR(errno);
n += tPutMapData(pWriter->aBuf[0] + n, mBlock); goto _err;
taosCalcChecksumAppend(0, pWriter->aBuf[0], size); }
ASSERT(n + sizeof(TSCKSUM) == size);
// write // read
n = taosWriteFile(pWriter->pHeadFD, pWriter->aBuf[0], size); n = taosReadFile(pReader->aLastFD[iSst], 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;
} }
// update // check
pBlockIdx->offset = pHeadFile->size; if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
pBlockIdx->size = size; code = TSDB_CODE_FILE_CORRUPTED;
pHeadFile->size += size; goto _err;
}
tsdbTrace("vgId:%d, write block, file ID:%d commit ID:%d suid:%" PRId64 " uid:%" PRId64 " offset:%" PRId64 // decode
" size:%" PRId64 " nItem:%d", n = 0;
TD_VID(pWriter->pTsdb->pVnode), pWriter->wSet.fid, pHeadFile->commitID, pBlockIdx->suid, pBlockIdx->uid, n = tGetU32(pReader->aBuf[0] + n, &delimiter);
pBlockIdx->offset, pBlockIdx->size, mBlock->nItem); ASSERT(delimiter == TSDB_FILE_DLMT);
while (n < size - sizeof(TSCKSUM)) {
SSstBlk blockl;
n += tGetSstBlk(pReader->aBuf[0] + n, &blockl);
if (taosArrayPush(aSstBlk, &blockl) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
}
ASSERT(n + sizeof(TSCKSUM) == size);
_exit:
return code; return code;
_err: _err:
tsdbError("vgId:%d, write block failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d read blockl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk) { int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *mBlock) {
int32_t code = 0; int32_t code = 0;
SSstFile *pSstFile = &pWriter->fSst[pWriter->wSet.nSstF - 1]; int64_t offset = pBlockIdx->offset;
int64_t size; int64_t size = pBlockIdx->size;
int64_t n; int64_t n;
int64_t tn;
// 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); code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err; if (code) goto _err;
// encode // seek
n = 0; if (taosLSeekFile(pReader->pHeadFD, offset, SEEK_SET) < 0) {
n += tPutU32(pWriter->aBuf[0] + n, TSDB_FILE_DLMT); code = TAOS_SYSTEM_ERROR(errno);
for (int32_t iBlockL = 0; iBlockL < taosArrayGetSize(aSstBlk); iBlockL++) { goto _err;
n += tPutSstBlk(pWriter->aBuf[0] + n, taosArrayGet(aSstBlk, iBlockL));
} }
taosCalcChecksumAppend(0, pWriter->aBuf[0], size);
ASSERT(n + sizeof(TSCKSUM) == size);
// write // read
n = taosWriteFile(pWriter->pLastFD, pWriter->aBuf[0], 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;
} }
// update // check
pSstFile->offset = pSstFile->size; if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
pSstFile->size += size; code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
_exit: }
tsdbTrace("vgId:%d tsdb write blockl, loffset:%" PRId64 " size:%" PRId64, TD_VID(pWriter->pTsdb->pVnode),
pSstFile->offset, size);
return code;
_err: // decode
tsdbError("vgId:%d tsdb write blockl failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); n = 0;
return code;
}
static void tsdbUpdateBlockInfo(SBlockData *pBlockData, SDataBlk *pDataBlk) { uint32_t delimiter;
for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) { n += tGetU32(pReader->aBuf[0] + n, &delimiter);
TSDBKEY key = {.ts = pBlockData->aTSKEY[iRow], .version = pBlockData->aVersion[iRow]}; ASSERT(delimiter == TSDB_FILE_DLMT);
if (iRow == 0) { tn = tGetMapData(pReader->aBuf[0] + n, mBlock);
if (tsdbKeyCmprFn(&pDataBlk->minKey, &key) > 0) { if (tn < 0) {
pDataBlk->minKey = key; code = TSDB_CODE_OUT_OF_MEMORY;
} goto _err;
} else { }
if (pBlockData->aTSKEY[iRow] == pBlockData->aTSKEY[iRow - 1]) { n += tn;
pDataBlk->hasDup = 1; ASSERT(n + sizeof(TSCKSUM) == size);
}
}
if (iRow == pBlockData->nRow - 1 && tsdbKeyCmprFn(&pDataBlk->maxKey, &key) < 0) { return code;
pDataBlk->maxKey = key;
}
pDataBlk->minVer = TMIN(pDataBlk->minVer, key.version); _err:
pDataBlk->maxVer = TMAX(pDataBlk->maxVer, key.version); tsdbError("vgId:%d, read block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
} return code;
pDataBlk->nRow += pBlockData->nRow;
} }
static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, SSmaInfo *pSmaInfo) { int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pDataBlk, SArray *aColumnDataAgg) {
int32_t code = 0; int32_t code = 0;
SSmaInfo *pSmaInfo = &pDataBlk->smaInfo;
pSmaInfo->offset = 0;
pSmaInfo->size = 0;
// encode ASSERT(pSmaInfo->size > 0);
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; taosArrayClear(aColumnDataAgg);
SColumnDataAgg sma; // alloc
tsdbCalcColDataSMA(pColData, &sma); int32_t size = pSmaInfo->size + sizeof(TSCKSUM);
code = tRealloc(&pReader->aBuf[0], size);
if (code) goto _err;
code = tRealloc(&pWriter->aBuf[0], pSmaInfo->size + tPutColumnDataAgg(NULL, &sma)); // seek
if (code) goto _err; int64_t n = taosLSeekFile(pReader->pSmaFD, pSmaInfo->offset, SEEK_SET);
pSmaInfo->size += tPutColumnDataAgg(pWriter->aBuf[0] + pSmaInfo->size, &sma); if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} else if (n < pSmaInfo->offset) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
} }
// write // read
if (pSmaInfo->size) { n = taosReadFile(pReader->pSmaFD, pReader->aBuf[0], size);
int32_t size = pSmaInfo->size + sizeof(TSCKSUM); if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
code = tRealloc(&pWriter->aBuf[0], size); // check
if (code) goto _err; if (!taosCheckChecksumWhole(pReader->aBuf[0], size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _err;
}
taosCalcChecksumAppend(0, pWriter->aBuf[0], size); // decode
n = 0;
while (n < pSmaInfo->size) {
SColumnDataAgg sma;
int64_t n = taosWriteFile(pWriter->pSmaFD, pWriter->aBuf[0], size); n += tGetColumnDataAgg(pReader->aBuf[0] + n, &sma);
if (n < 0) { if (taosArrayPush(aColumnDataAgg, &sma) == NULL) {
code = TAOS_SYSTEM_ERROR(errno); code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
pSmaInfo->offset = pWriter->fSma.size;
pWriter->fSma.size += 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 tsdb read block sma 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, static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, int8_t fromLast,
int8_t cmprAlg, int8_t toLast) { SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
ASSERT(pBlockData->nRow > 0); tBlockDataClear(pBlockData);
TdFilePtr pFD = fromLast ? pReader->aLastFD[0] : pReader->pDataFD; // (todo)
// uid + version + tskey
code = tsdbReadAndCheck(pFD, pBlkInfo->offset, &pReader->aBuf[0], pBlkInfo->szKey, 1);
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->uid == hdr.uid);
pBlockData->nRow = hdr.nRow;
// uid
if (hdr.uid == 0) {
ASSERT(hdr.szUid);
code = tsdbDecmprData(p, hdr.szUid, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aUid,
sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err;
} else {
ASSERT(!hdr.szUid);
}
p += hdr.szUid;
pBlkInfo->offset = toLast ? pWriter->fSst[pWriter->wSet.nSstF - 1].size : pWriter->fData.size; // version
pBlkInfo->szBlock = 0; code = tsdbDecmprData(p, hdr.szVer, TSDB_DATA_TYPE_BIGINT, hdr.cmprAlg, (uint8_t **)&pBlockData->aVersion,
pBlkInfo->szKey = 0; sizeof(int64_t) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err;
p += hdr.szVer;
int32_t aBufN[4] = {0}; // TSKEY
code = tCmprBlockData(pBlockData, cmprAlg, NULL, NULL, pWriter->aBuf, aBufN); code = tsdbDecmprData(p, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
sizeof(TSKEY) * hdr.nRow, &pReader->aBuf[1]);
if (code) goto _err; if (code) goto _err;
p += hdr.szKey;
// write ================= ASSERT(p - pReader->aBuf[0] == pBlkInfo->szKey - sizeof(TSCKSUM));
TdFilePtr pFD = toLast ? pWriter->pLastFD : pWriter->pDataFD;
pBlkInfo->szKey = aBufN[3] + aBufN[2]; // read and decode columns
pBlkInfo->szBlock = aBufN[0] + aBufN[1] + aBufN[2] + aBufN[3]; if (taosArrayGetSize(pBlockData->aIdx) == 0) goto _exit;
int64_t n = taosWriteFile(pFD, pWriter->aBuf[3], aBufN[3]); if (hdr.szBlkCol > 0) {
if (n < 0) { int64_t offset = pBlkInfo->offset + pBlkInfo->szKey;
code = TAOS_SYSTEM_ERROR(errno); code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[0], hdr.szBlkCol + sizeof(TSCKSUM), 1);
goto _err; if (code) goto _err;
} }
n = taosWriteFile(pFD, pWriter->aBuf[2], aBufN[2]); SBlockCol blockCol = {.cid = 0};
if (n < 0) { SBlockCol *pBlockCol = &blockCol;
code = TAOS_SYSTEM_ERROR(errno); int32_t n = 0;
goto _err;
}
if (aBufN[1]) { for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
n = taosWriteFile(pFD, pWriter->aBuf[1], aBufN[1]); SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
}
if (aBufN[0]) { while (pBlockCol && pBlockCol->cid < pColData->cid) {
n = taosWriteFile(pFD, pWriter->aBuf[0], aBufN[0]); if (n < hdr.szBlkCol) {
if (n < 0) { n += tGetBlockCol(pReader->aBuf[0] + n, pBlockCol);
code = TAOS_SYSTEM_ERROR(errno); } else {
goto _err; ASSERT(n == hdr.szBlkCol);
pBlockCol = NULL;
}
} }
}
// update info if (pBlockCol == NULL || pBlockCol->cid > pColData->cid) {
if (toLast) { // add a lot of NONE
pWriter->fSst[pWriter->wSet.nSstF - 1].size += pBlkInfo->szBlock; for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
} else { code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
pWriter->fData.size += pBlkInfo->szBlock; if (code) goto _err;
} }
} else {
ASSERT(pBlockCol->type == pColData->type);
ASSERT(pBlockCol->flag && pBlockCol->flag != HAS_NONE);
// ================= SMA ==================== if (pBlockCol->flag == HAS_NULL) {
if (pSmaInfo) { // add a lot of NULL
code = tsdbWriteBlockSma(pWriter, pBlockData, pSmaInfo); for (int32_t iRow = 0; iRow < hdr.nRow; iRow++) {
if (code) goto _err; code = tColDataAppendValue(pColData, &COL_VAL_NULL(pBlockCol->cid, pBlockCol->type));
if (code) goto _err;
}
} else {
// decode from binary
int64_t offset = pBlkInfo->offset + pBlkInfo->szKey + hdr.szBlkCol + sizeof(TSCKSUM) + pBlockCol->offset;
int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
code = tsdbReadAndCheck(pFD, offset, &pReader->aBuf[1], size, 0);
if (code) goto _err;
code = tsdbDecmprColData(pReader->aBuf[1], pBlockCol, hdr.cmprAlg, hdr.nRow, pColData, &pReader->aBuf[2]);
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 tsdb write block data failed since %s", TD_VID(pWriter->pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d tsdb read block data impl failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
int32_t tsdbDFileSetCopy(STsdb *pTsdb, SDFileSet *pSetFrom, SDFileSet *pSetTo) { int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) {
int32_t code = 0; 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 code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], 0, pBlockData);
tsdbHeadFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pHeadF, fNameFrom); if (code) goto _err;
tsdbHeadFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pHeadF, fNameTo);
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); if (pDataBlk->nSubBlock > 1) {
if (pOutFD == NULL) { SBlockData bData1;
code = TAOS_SYSTEM_ERROR(errno); SBlockData bData2;
goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ); // create
if (PInFD == NULL) { code = tBlockDataCreate(&bData1);
code = TAOS_SYSTEM_ERROR(errno); if (code) goto _err;
goto _err; code = tBlockDataCreate(&bData2);
} if (code) goto _err;
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pHeadF->size); // init
if (n < 0) { tBlockDataInitEx(&bData1, pBlockData);
code = TAOS_SYSTEM_ERROR(errno); tBlockDataInitEx(&bData2, pBlockData);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// data for (int32_t iSubBlock = 1; iSubBlock < pDataBlk->nSubBlock; iSubBlock++) {
tsdbDataFileName(pTsdb, pSetFrom->diskId, pSetFrom->fid, pSetFrom->pDataF, fNameFrom); code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[iSubBlock], 0, &bData1);
tsdbDataFileName(pTsdb, pSetTo->diskId, pSetTo->fid, pSetTo->pDataF, fNameTo); if (code) {
tBlockDataDestroy(&bData1, 1);
tBlockDataDestroy(&bData2, 1);
goto _err;
}
pOutFD = taosOpenFile(fNameTo, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC); code = tBlockDataCopy(pBlockData, &bData2);
if (pOutFD == NULL) { if (code) {
code = TAOS_SYSTEM_ERROR(errno); tBlockDataDestroy(&bData1, 1);
goto _err; tBlockDataDestroy(&bData2, 1);
} goto _err;
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ); code = tBlockDataMerge(&bData1, &bData2, pBlockData);
if (PInFD == NULL) { if (code) {
code = TAOS_SYSTEM_ERROR(errno); tBlockDataDestroy(&bData1, 1);
goto _err; tBlockDataDestroy(&bData2, 1);
} goto _err;
}
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pDataF->size); tBlockDataDestroy(&bData1, 1);
if (n < 0) { tBlockDataDestroy(&bData2, 1);
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
} }
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sst return code;
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); _err:
if (pOutFD == NULL) { tsdbError("vgId:%d tsdb read data block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
code = TAOS_SYSTEM_ERROR(errno); return code;
goto _err; }
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ); int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
if (PInFD == NULL) { int32_t code = 0;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->aSstF[0]->size); code = tsdbReadBlockDataImpl(pReader, &pSstBlk->bInfo, 1, pBlockData);
if (n < 0) { if (code) goto _err;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
// sma return code;
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); _err:
if (pOutFD == NULL) { tsdbError("vgId:%d tsdb read last block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
code = TAOS_SYSTEM_ERROR(errno); return code;
goto _err; }
}
PInFD = taosOpenFile(fNameFrom, TD_FILE_READ); int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData) {
if (PInFD == NULL) { int32_t code = 0;
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
n = taosFSendFile(pOutFD, PInFD, 0, pSetFrom->pSmaF->size); // read
if (n < 0) { code = tsdbReadAndCheck(pReader->aLastFD[iSst], pSstBlk->bInfo.offset, &pReader->aBuf[0], pSstBlk->bInfo.szBlock, 0);
code = TAOS_SYSTEM_ERROR(errno); if (code) goto _exit;
goto _err;
}
taosCloseFile(&pOutFD);
taosCloseFile(&PInFD);
return code; // decmpr
code = tDecmprBlockData(pReader->aBuf[0], pSstBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]);
if (code) goto _exit;
_err: _exit:
tsdbError("vgId:%d, tsdb DFileSet copy failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
...@@ -1558,7 +1534,6 @@ _err: ...@@ -1558,7 +1534,6 @@ _err:
tsdbError("vgId:%d, update del file hdr 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 ==================================================== // SDelFReader ====================================================
struct SDelFReader { struct SDelFReader {
STsdb *pTsdb; STsdb *pTsdb;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或