Skip to content

T
TDengine

dgr6878 / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 8
Star 0
Fork 0
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 9845bd94 编写于 作者: L liuyq-617
浏览文件
操作

Merge branches 'test/jenkins' and 'feature/TD-1925_new' of... 

Merge branches 'test/jenkins' and 'feature/TD-1925_new' of github.com:taosdata/TDengine into feature/TD-1925_new
上级 0d46d2b2 e5dce6e1
显示空白变更内容
内联 并排
Showing with 105 addition and 27 deletion
src/os/src/detail/osFile.c
浏览文件 @ 9845bd94
...@@ -142,6 +142,8 @@ int64_t taosCopy(char *from, char *to) { ...@@ -142,6 +142,8 @@ int64_t taosCopy(char *from, char *to) {
if (bytes < sizeof(buffer)) break; if (bytes < sizeof(buffer)) break;
} }
fsync(fidto);
close(fidfrom); close(fidfrom);
close(fidto); close(fidto);
return size; return size;
......
src/plugins/http/src/httpRestJson.c
浏览文件 @ 9845bd94
...@@ -35,7 +35,7 @@ void restBuildSqlAffectRowsJson(HttpContext *pContext, HttpSqlCmd *cmd, int32_t ...@@ -35,7 +35,7 @@ void restBuildSqlAffectRowsJson(HttpContext *pContext, HttpSqlCmd *cmd, int32_t
// data row array end // data row array end
httpJsonToken(jsonBuf, JsonArrEnd); httpJsonToken(jsonBuf, JsonArrEnd);
cmd->numOfRows = affect_rows; cmd->numOfRows = 1;
} }
void restStartSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result) { void restStartSqlJson(HttpContext *pContext, HttpSqlCmd *cmd, TAOS_RES *result) {
......
src/tsdb/inc/tsdbFile.h
浏览文件 @ 9845bd94
...@@ -50,7 +50,7 @@ typedef struct { ...@@ -50,7 +50,7 @@ typedef struct {
} SMFile; } SMFile;
void tsdbInitMFile(SMFile* pMFile, SDiskID did, int vid, uint32_t ver); void tsdbInitMFile(SMFile* pMFile, SDiskID did, int vid, uint32_t ver);
void tsdbInitMFileEx(SMFile* pMFile, SMFile* pOMFile); void tsdbInitMFileEx(SMFile* pMFile, const SMFile* pOMFile);
int tsdbEncodeSMFile(void** buf, SMFile* pMFile); int tsdbEncodeSMFile(void** buf, SMFile* pMFile);
void* tsdbDecodeSMFile(void* buf, SMFile* pMFile); void* tsdbDecodeSMFile(void* buf, SMFile* pMFile);
int tsdbEncodeSMFileEx(void** buf, SMFile* pMFile); int tsdbEncodeSMFileEx(void** buf, SMFile* pMFile);
......
src/tsdb/src/tsdbCommit.c
浏览文件 @ 9845bd94
...@@ -52,7 +52,7 @@ static int tsdbCommitMeta(STsdbRepo *pRepo); ...@@ -52,7 +52,7 @@ static int tsdbCommitMeta(STsdbRepo *pRepo);
static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen); static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen);
static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid); static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid);
static int tsdbCommitTSData(STsdbRepo *pRepo); static int tsdbCommitTSData(STsdbRepo *pRepo);
static int tsdbStartCommit(STsdbRepo *pRepo); static void tsdbStartCommit(STsdbRepo *pRepo);
static void tsdbEndCommit(STsdbRepo *pRepo, int eno); static void tsdbEndCommit(STsdbRepo *pRepo, int eno);
static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid); static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
static int tsdbCreateCommitIters(SCommitH *pCommith); static int tsdbCreateCommitIters(SCommitH *pCommith);
...@@ -84,10 +84,7 @@ static int tsdbApplyRtn(STsdbRepo *pRepo); ...@@ -84,10 +84,7 @@ static int tsdbApplyRtn(STsdbRepo *pRepo);
static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn); static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn);
void *tsdbCommitData(STsdbRepo *pRepo) { void *tsdbCommitData(STsdbRepo *pRepo) {
if (tsdbStartCommit(pRepo) < 0) { tsdbStartCommit(pRepo);
tsdbError("vgId:%d failed to commit data while startting to commit since %s", REPO_ID(pRepo), tstrerror(terrno));
goto _err;
}
// Commit to update meta file // Commit to update meta file
if (tsdbCommitMeta(pRepo) < 0) { if (tsdbCommitMeta(pRepo) < 0) {
...@@ -138,11 +135,15 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { ...@@ -138,11 +135,15 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
tsdbInitMFile(&mf, did, REPO_ID(pRepo), FS_TXN_VERSION(REPO_FS(pRepo))); tsdbInitMFile(&mf, did, REPO_ID(pRepo), FS_TXN_VERSION(REPO_FS(pRepo)));
if (tsdbCreateMFile(&mf, true) < 0) { if (tsdbCreateMFile(&mf, true) < 0) {
tsdbError("vgId:%d failed to create META file since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1; return -1;
} }
tsdbInfo("vgId:%d meta file %s is created to commit", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(&mf));
} else { } else {
tsdbInitMFileEx(&mf, pOMFile); tsdbInitMFileEx(&mf, pOMFile);
if (tsdbOpenMFile(&mf, O_WRONLY) < 0) { if (tsdbOpenMFile(&mf, O_WRONLY) < 0) {
tsdbError("vgId:%d failed to open META file since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1; return -1;
} }
} }
...@@ -154,12 +155,20 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { ...@@ -154,12 +155,20 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
if (pAct->act == TSDB_UPDATE_META) { if (pAct->act == TSDB_UPDATE_META) {
pCont = (SActCont *)POINTER_SHIFT(pAct, sizeof(SActObj)); pCont = (SActCont *)POINTER_SHIFT(pAct, sizeof(SActObj));
if (tsdbUpdateMetaRecord(pfs, &mf, pAct->uid, (void *)(pCont->cont), pCont->len) < 0) { if (tsdbUpdateMetaRecord(pfs, &mf, pAct->uid, (void *)(pCont->cont), pCont->len) < 0) {
tsdbError("vgId:%d failed to update META record, uid %" PRIu64 " since %s", REPO_ID(pRepo), pAct->uid,
tstrerror(terrno));
tsdbCloseMFile(&mf); tsdbCloseMFile(&mf);
tsdbApplyMFileChange(&mf, pOMFile);
// TODO: need to reload metaCache
return -1; return -1;
} }
} else if (pAct->act == TSDB_DROP_META) { } else if (pAct->act == TSDB_DROP_META) {
if (tsdbDropMetaRecord(pfs, &mf, pAct->uid) < 0) { if (tsdbDropMetaRecord(pfs, &mf, pAct->uid) < 0) {
tsdbError("vgId:%d failed to drop META record, uid %" PRIu64 " since %s", REPO_ID(pRepo), pAct->uid,
tstrerror(terrno));
tsdbCloseMFile(&mf); tsdbCloseMFile(&mf);
tsdbApplyMFileChange(&mf, pOMFile);
// TODO: need to reload metaCache
return -1; return -1;
} }
} else { } else {
...@@ -168,6 +177,9 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { ...@@ -168,6 +177,9 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) {
} }
if (tsdbUpdateMFileHeader(&mf) < 0) { if (tsdbUpdateMFileHeader(&mf) < 0) {
tsdbError("vgId:%d failed to update META file header since %s, revert it", REPO_ID(pRepo), tstrerror(terrno));
tsdbApplyMFileChange(&mf, pOMFile);
// TODO: need to reload metaCache
return -1; return -1;
} }
...@@ -238,7 +250,7 @@ static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void ...@@ -238,7 +250,7 @@ static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void
tsdbUpdateMFileMagic(pMFile, POINTER_SHIFT(cont, contLen - sizeof(TSCKSUM))); tsdbUpdateMFileMagic(pMFile, POINTER_SHIFT(cont, contLen - sizeof(TSCKSUM)));
SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)&uid, sizeof(uid)); SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)&uid, sizeof(uid));
if (pRecord != NULL) { if (pRecord != NULL) {
pMFile->info.tombSize += pRecord->size; pMFile->info.tombSize += (pRecord->size + sizeof(SKVRecord));
} else { } else {
pMFile->info.nRecords++; pMFile->info.nRecords++;
} }
...@@ -253,7 +265,7 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) { ...@@ -253,7 +265,7 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)(&uid), sizeof(uid)); SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)(&uid), sizeof(uid));
if (pRecord == NULL) { if (pRecord == NULL) {
tsdbError("failed to drop KV store record with key %" PRIu64 " since not find", uid); tsdbError("failed to drop META record with key %" PRIu64 " since not find", uid);
return -1; return -1;
} }
...@@ -264,11 +276,11 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) { ...@@ -264,11 +276,11 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) {
void *pBuf = buf; void *pBuf = buf;
tsdbEncodeKVRecord(&pBuf, &rInfo); tsdbEncodeKVRecord(&pBuf, &rInfo);
if (tsdbAppendMFile(pMFile, buf, POINTER_DISTANCE(pBuf, buf), NULL) < 0) { if (tsdbAppendMFile(pMFile, buf, sizeof(SKVRecord), NULL) < 0) {
return -1; return -1;
} }
pMFile->info.magic = taosCalcChecksum(pMFile->info.magic, (uint8_t *)buf, (uint32_t)POINTER_DISTANCE(pBuf, buf)); pMFile->info.magic = taosCalcChecksum(pMFile->info.magic, (uint8_t *)buf, sizeof(SKVRecord));
pMFile->info.nDels++; pMFile->info.nDels++;
pMFile->info.nRecords--; pMFile->info.nRecords--;
pMFile->info.tombSize += (rInfo.size + sizeof(SKVRecord) * 2); pMFile->info.tombSize += (rInfo.size + sizeof(SKVRecord) * 2);
...@@ -302,7 +314,12 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) { ...@@ -302,7 +314,12 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
// Skip expired memory data and expired FSET // Skip expired memory data and expired FSET
tsdbSeekCommitIter(&commith, commith.rtn.minKey); tsdbSeekCommitIter(&commith, commith.rtn.minKey);
while ((pSet = tsdbFSIterNext(&(commith.fsIter)))) { while ((pSet = tsdbFSIterNext(&(commith.fsIter)))) {
if (pSet->fid >= commith.rtn.minFid) break; if (pSet->fid < commith.rtn.minFid) {
tsdbInfo("vgId:%d FSET %d on level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
} else {
break;
}
} }
// Loop to commit to each file // Loop to commit to each file
...@@ -349,7 +366,7 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) { ...@@ -349,7 +366,7 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
return 0; return 0;
} }
static int tsdbStartCommit(STsdbRepo *pRepo) { static void tsdbStartCommit(STsdbRepo *pRepo) {
SMemTable *pMem = pRepo->imem; SMemTable *pMem = pRepo->imem;
ASSERT(pMem->numOfRows > 0 || listNEles(pMem->actList) > 0); ASSERT(pMem->numOfRows > 0 || listNEles(pMem->actList) > 0);
...@@ -360,7 +377,6 @@ static int tsdbStartCommit(STsdbRepo *pRepo) { ...@@ -360,7 +377,6 @@ static int tsdbStartCommit(STsdbRepo *pRepo) {
tsdbStartFSTxn(pRepo, pMem->pointsAdd, pMem->storageAdd); tsdbStartFSTxn(pRepo, pMem->pointsAdd, pMem->storageAdd);
pRepo->code = TSDB_CODE_SUCCESS; pRepo->code = TSDB_CODE_SUCCESS;
return 0;
} }
static void tsdbEndCommit(STsdbRepo *pRepo, int eno) { static void tsdbEndCommit(STsdbRepo *pRepo, int eno) {
...@@ -413,14 +429,18 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) { ...@@ -413,14 +429,18 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
if (pIter->pTable == NULL) continue; if (pIter->pTable == NULL) continue;
if (tsdbCommitToTable(pCommith, tid) < 0) { if (tsdbCommitToTable(pCommith, tid) < 0) {
// TODO: revert the file change
tsdbCloseCommitFile(pCommith, true); tsdbCloseCommitFile(pCommith, true);
// revert the file change
tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
return -1; return -1;
} }
} }
if (tsdbWriteBlockIdx(pCommith) < 0) { if (tsdbWriteBlockIdx(pCommith) < 0) {
tsdbError("vgId:%d failed to write SBlockIdx part to FSET %d since %s", REPO_ID(pRepo), fid, tstrerror(terrno));
tsdbCloseCommitFile(pCommith, true); tsdbCloseCommitFile(pCommith, true);
// revert the file change
tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
return -1; return -1;
} }
...@@ -674,7 +694,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) { ...@@ -674,7 +694,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
TSDB_RUNLOCK_TABLE(pIter->pTable); TSDB_RUNLOCK_TABLE(pIter->pTable);
if (tsdbWriteBlockInfo(pCommith) < 0) return -1; if (tsdbWriteBlockInfo(pCommith) < 0) {
tsdbError("vgId:%d failed to write SBlockInfo part into file %s since %s", TSDB_COMMIT_REPO_ID(pCommith),
TSDB_FILE_FULL_NAME(TSDB_COMMIT_HEAD_FILE(pCommith)), tstrerror(terrno));
return -1;
}
return 0; return 0;
} }
...@@ -926,6 +950,8 @@ static int tsdbWriteBlockIdx(SCommitH *pCommih) { ...@@ -926,6 +950,8 @@ static int tsdbWriteBlockIdx(SCommitH *pCommih) {
if (nidx <= 0) { if (nidx <= 0) {
// All data are deleted // All data are deleted
pHeadf->info.offset = 0;
pHeadf->info.len = 0;
return 0; return 0;
} }
...@@ -1227,7 +1253,6 @@ static void tsdbResetCommitFile(SCommitH *pCommith) { ...@@ -1227,7 +1253,6 @@ static void tsdbResetCommitFile(SCommitH *pCommith) {
} }
static void tsdbResetCommitTable(SCommitH *pCommith) { static void tsdbResetCommitTable(SCommitH *pCommith) {
tdResetDataCols(pCommith->pDataCols);
taosArrayClear(pCommith->aSubBlk); taosArrayClear(pCommith->aSubBlk);
taosArrayClear(pCommith->aSupBlk); taosArrayClear(pCommith->aSupBlk);
pCommith->pTable = NULL; pCommith->pTable = NULL;
...@@ -1256,6 +1281,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1256,6 +1281,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbCloseAndUnsetFSet(&(pCommith->readh)); tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1; return -1;
} }
tsdbDebug("vgId:%d FSET %d at level %d disk id %d is opened to read to commit", REPO_ID(pRepo), TSDB_FSET_FID(pSet),
TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
} else { } else {
pCommith->isRFileSet = false; pCommith->isRFileSet = false;
} }
...@@ -1266,6 +1294,8 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1266,6 +1294,8 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbInitDFileSet(pWSet, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo))); tsdbInitDFileSet(pWSet, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)));
if (tsdbCreateDFileSet(pWSet, true) < 0) { if (tsdbCreateDFileSet(pWSet, true) < 0) {
tsdbError("vgId:%d failed to create FSET %d at level %d disk id %d since %s", REPO_ID(pRepo),
TSDB_FSET_FID(pWSet), TSDB_FSET_LEVEL(pWSet), TSDB_FSET_ID(pWSet), tstrerror(terrno));
if (pCommith->isRFileSet) { if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh)); tsdbCloseAndUnsetFSet(&(pCommith->readh));
} }
...@@ -1274,6 +1304,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1274,6 +1304,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith->isDFileSame = false; pCommith->isDFileSame = false;
pCommith->isLFileSame = false; pCommith->isLFileSame = false;
tsdbDebug("vgId:%d FSET %d at level %d disk id %d is created to commit", REPO_ID(pRepo), TSDB_FSET_FID(pWSet),
TSDB_FSET_LEVEL(pWSet), TSDB_FSET_ID(pWSet));
} else { } else {
did.level = TSDB_FSET_LEVEL(pSet); did.level = TSDB_FSET_LEVEL(pSet);
did.id = TSDB_FSET_ID(pSet); did.id = TSDB_FSET_ID(pSet);
...@@ -1285,6 +1318,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1285,6 +1318,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile *pWHeadf = TSDB_COMMIT_HEAD_FILE(pCommith); SDFile *pWHeadf = TSDB_COMMIT_HEAD_FILE(pCommith);
tsdbInitDFile(pWHeadf, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_HEAD); tsdbInitDFile(pWHeadf, did, REPO_ID(pRepo), fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_HEAD);
if (tsdbCreateDFile(pWHeadf, true) < 0) { if (tsdbCreateDFile(pWHeadf, true) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWHeadf),
tstrerror(terrno));
if (pCommith->isRFileSet) { if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh)); tsdbCloseAndUnsetFSet(&(pCommith->readh));
return -1; return -1;
...@@ -1296,7 +1332,10 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1296,7 +1332,10 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile *pWDataf = TSDB_COMMIT_DATA_FILE(pCommith); SDFile *pWDataf = TSDB_COMMIT_DATA_FILE(pCommith);
tsdbInitDFileEx(pWDataf, pRDataf); tsdbInitDFileEx(pWDataf, pRDataf);
if (tsdbOpenDFile(pWDataf, O_WRONLY) < 0) { if (tsdbOpenDFile(pWDataf, O_WRONLY) < 0) {
tsdbCloseDFile(pWHeadf); tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWDataf),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet);
tsdbRemoveDFile(pWHeadf); tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) { if (pCommith->isRFileSet) {
tsdbCloseAndUnsetFSet(&(pCommith->readh)); tsdbCloseAndUnsetFSet(&(pCommith->readh));
...@@ -1313,6 +1352,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1313,6 +1352,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith->isLFileSame = true; pCommith->isLFileSame = true;
if (tsdbOpenDFile(pWLastf, O_WRONLY) < 0) { if (tsdbOpenDFile(pWLastf, O_WRONLY) < 0) {
tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWLastf),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet); tsdbCloseDFileSet(pWSet);
tsdbRemoveDFile(pWHeadf); tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) { if (pCommith->isRFileSet) {
...@@ -1325,6 +1367,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid ...@@ -1325,6 +1367,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith->isLFileSame = false; pCommith->isLFileSame = false;
if (tsdbCreateDFile(pWLastf, true) < 0) { if (tsdbCreateDFile(pWLastf, true) < 0) {
tsdbError("vgId:%d failed to create file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWLastf),
tstrerror(terrno));
tsdbCloseDFileSet(pWSet); tsdbCloseDFileSet(pWSet);
tsdbRemoveDFile(pWHeadf); tsdbRemoveDFile(pWHeadf);
if (pCommith->isRFileSet) { if (pCommith->isRFileSet) {
...@@ -1360,7 +1405,7 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p ...@@ -1360,7 +1405,7 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
if (pBlock->last) { if (pBlock->last) {
if (pCommith->isLFileSame && mergeRows < pCfg->minRowsPerFileBlock) return true; if (pCommith->isLFileSame && mergeRows < pCfg->minRowsPerFileBlock) return true;
} else { } else {
if (mergeRows < pCfg->maxRowsPerFileBlock) return true; if (pCommith->isDFileSame && mergeRows <= pCfg->maxRowsPerFileBlock) return true;
} }
} }
...@@ -1373,12 +1418,16 @@ static int tsdbApplyRtn(STsdbRepo *pRepo) { ...@@ -1373,12 +1418,16 @@ static int tsdbApplyRtn(STsdbRepo *pRepo) {
STsdbFS * pfs = REPO_FS(pRepo); STsdbFS * pfs = REPO_FS(pRepo);
SDFileSet *pSet; SDFileSet *pSet;
// Get retentioni snapshot // Get retention snapshot
tsdbGetRtnSnap(pRepo, &rtn); tsdbGetRtnSnap(pRepo, &rtn);
tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD); tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
while ((pSet = tsdbFSIterNext(&fsiter))) { while ((pSet = tsdbFSIterNext(&fsiter))) {
if (pSet->fid < rtn.minFid) continue; if (pSet->fid < rtn.minFid) {
tsdbInfo("vgId:%d FSET %d at level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid,
TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet));
continue;
}
if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) { if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) {
return -1; return -1;
...@@ -1392,10 +1441,13 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) { ...@@ -1392,10 +1441,13 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
SDiskID did; SDiskID did;
SDFileSet nSet; SDFileSet nSet;
STsdbFS * pfs = REPO_FS(pRepo); STsdbFS * pfs = REPO_FS(pRepo);
int level;
ASSERT(pSet->fid >= pRtn->minFid); ASSERT(pSet->fid >= pRtn->minFid);
tfsAllocDisk(tsdbGetFidLevel(pSet->fid, pRtn), &(did.level), &(did.id)); level = tsdbGetFidLevel(pSet->fid, pRtn);
tfsAllocDisk(level, &(did.level), &(did.id));
if (did.level == TFS_UNDECIDED_LEVEL) { if (did.level == TFS_UNDECIDED_LEVEL) {
terrno = TSDB_CODE_TDB_NO_AVAIL_DISK; terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
return -1; return -1;
...@@ -1406,12 +1458,17 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) { ...@@ -1406,12 +1458,17 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs)); tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs));
if (tsdbCopyDFileSet(pSet, &nSet) < 0) { if (tsdbCopyDFileSet(pSet, &nSet) < 0) {
tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid,
TSDB_FSET_LEVEL(pSet), did.level, tstrerror(terrno));
return -1; return -1;
} }
if (tsdbUpdateDFileSet(pfs, &nSet) < 0) { if (tsdbUpdateDFileSet(pfs, &nSet) < 0) {
return -1; return -1;
} }
tsdbInfo("vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d", REPO_ID(pRepo), pSet->fid,
TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet), did.level, did.id);
} else { } else {
// On a correct level // On a correct level
if (tsdbUpdateDFileSet(pfs, pSet) < 0) { if (tsdbUpdateDFileSet(pfs, pSet) < 0) {
......
src/tsdb/src/tsdbFS.c
浏览文件 @ 9845bd94
...@@ -701,6 +701,8 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) { ...@@ -701,6 +701,8 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
int64_t maxBufSize = 0; int64_t maxBufSize = 0;
SMFInfo minfo; SMFInfo minfo;
taosHashEmpty(pfs->metaCache);
// No meta file, just return // No meta file, just return
if (pfs->cstatus->pmf == NULL) return 0; if (pfs->cstatus->pmf == NULL) return 0;
...@@ -718,6 +720,12 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) { ...@@ -718,6 +720,12 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) {
while (true) { while (true) {
int64_t tsize = tsdbReadMFile(pMFile, tbuf, sizeof(SKVRecord)); int64_t tsize = tsdbReadMFile(pMFile, tbuf, sizeof(SKVRecord));
if (tsize == 0) break; if (tsize == 0) break;
if (tsize < 0) {
tsdbError("vgId:%d failed to read META file since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1;
}
if (tsize < sizeof(SKVRecord)) { if (tsize < sizeof(SKVRecord)) {
tsdbError("vgId:%d failed to read %" PRIzu " bytes from file %s", REPO_ID(pRepo), sizeof(SKVRecord), tsdbError("vgId:%d failed to read %" PRIzu " bytes from file %s", REPO_ID(pRepo), sizeof(SKVRecord),
TSDB_FILE_FULL_NAME(pMFile)); TSDB_FILE_FULL_NAME(pMFile));
......
src/tsdb/src/tsdbFile.c
浏览文件 @ 9845bd94
...@@ -42,7 +42,7 @@ void tsdbInitMFile(SMFile *pMFile, SDiskID did, int vid, uint32_t ver) { ...@@ -42,7 +42,7 @@ void tsdbInitMFile(SMFile *pMFile, SDiskID did, int vid, uint32_t ver) {
tfsInitFile(TSDB_FILE_F(pMFile), did.level, did.id, fname); tfsInitFile(TSDB_FILE_F(pMFile), did.level, did.id, fname);
} }
void tsdbInitMFileEx(SMFile *pMFile, SMFile *pOMFile) { void tsdbInitMFileEx(SMFile *pMFile, const SMFile *pOMFile) {
*pMFile = *pOMFile; *pMFile = *pOMFile;
TSDB_FILE_SET_CLOSED(pMFile); TSDB_FILE_SET_CLOSED(pMFile);
} }
......
src/vnode/src/vnodeCfg.c
浏览文件 @ 9845bd94
...@@ -34,6 +34,7 @@ static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) { ...@@ -34,6 +34,7 @@ static void vnodeLoadCfg(SVnodeObj *pVnode, SCreateVnodeMsg* vnodeMsg) {
pVnode->tsdbCfg.maxRowsPerFileBlock = vnodeMsg->cfg.maxRowsPerFileBlock; pVnode->tsdbCfg.maxRowsPerFileBlock = vnodeMsg->cfg.maxRowsPerFileBlock;
pVnode->tsdbCfg.precision = vnodeMsg->cfg.precision; pVnode->tsdbCfg.precision = vnodeMsg->cfg.precision;
pVnode->tsdbCfg.compression = vnodeMsg->cfg.compression; pVnode->tsdbCfg.compression = vnodeMsg->cfg.compression;
pVnode->tsdbCfg.update = vnodeMsg->cfg.update;
pVnode->tsdbCfg.cacheLastRow = vnodeMsg->cfg.cacheLastRow; pVnode->tsdbCfg.cacheLastRow = vnodeMsg->cfg.cacheLastRow;
pVnode->walCfg.walLevel = vnodeMsg->cfg.walLevel; pVnode->walCfg.walLevel = vnodeMsg->cfg.walLevel;
pVnode->walCfg.fsyncPeriod = vnodeMsg->cfg.fsyncPeriod; pVnode->walCfg.fsyncPeriod = vnodeMsg->cfg.fsyncPeriod;
...@@ -227,6 +228,15 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) { ...@@ -227,6 +228,15 @@ int32_t vnodeReadCfg(SVnodeObj *pVnode) {
} }
vnodeMsg.cfg.quorum = (int8_t)quorum->valueint; vnodeMsg.cfg.quorum = (int8_t)quorum->valueint;
cJSON *update = cJSON_GetObjectItem(root, "update");
if (!update || update->type != cJSON_Number) {
vError("vgId: %d, failed to read %s, update not found", pVnode->vgId, file);
vnodeMsg.cfg.update = 0;
vnodeMsg.cfg.vgCfgVersion = 0;
} else {
vnodeMsg.cfg.update = (int8_t)update->valueint;
}
cJSON *cacheLastRow = cJSON_GetObjectItem(root, "cacheLastRow"); cJSON *cacheLastRow = cJSON_GetObjectItem(root, "cacheLastRow");
if (!cacheLastRow || cacheLastRow->type != cJSON_Number) { if (!cacheLastRow || cacheLastRow->type != cJSON_Number) {
vError("vgId: %d, failed to read %s, cacheLastRow not found", pVnode->vgId, file); vError("vgId: %d, failed to read %s, cacheLastRow not found", pVnode->vgId, file);
...@@ -325,6 +335,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) { ...@@ -325,6 +335,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) {
len += snprintf(content + len, maxLen - len, " \"dbReplica\": %d,\n", pMsg->cfg.dbReplica); len += snprintf(content + len, maxLen - len, " \"dbReplica\": %d,\n", pMsg->cfg.dbReplica);
len += snprintf(content + len, maxLen - len, " \"wals\": %d,\n", pMsg->cfg.wals); len += snprintf(content + len, maxLen - len, " \"wals\": %d,\n", pMsg->cfg.wals);
len += snprintf(content + len, maxLen - len, " \"quorum\": %d,\n", pMsg->cfg.quorum); len += snprintf(content + len, maxLen - len, " \"quorum\": %d,\n", pMsg->cfg.quorum);
len += snprintf(content + len, maxLen - len, " \"update\": %d,\n", pMsg->cfg.update);
len += snprintf(content + len, maxLen - len, " \"cacheLastRow\": %d,\n", pMsg->cfg.cacheLastRow); len += snprintf(content + len, maxLen - len, " \"cacheLastRow\": %d,\n", pMsg->cfg.cacheLastRow);
len += snprintf(content + len, maxLen - len, " \"nodeInfos\": [{\n"); len += snprintf(content + len, maxLen - len, " \"nodeInfos\": [{\n");
for (int32_t i = 0; i < pMsg->cfg.vgReplica; i++) { for (int32_t i = 0; i < pMsg->cfg.vgReplica; i++) {
......
tests/script/general/http/restful_full.sim
浏览文件 @ 9845bd94
...@@ -94,7 +94,7 @@ endi ...@@ -94,7 +94,7 @@ endi
system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d 'create database d1' 127.0.0.1:7111/rest/sql system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d 'create database d1' 127.0.0.1:7111/rest/sql
print 12-> $system_content print 12-> $system_content
if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":0}@ then if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":1}@ then
return -1 return -1
endi endi
...@@ -160,7 +160,7 @@ endi ...@@ -160,7 +160,7 @@ endi
system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d ' create table d1.t1 (ts timestamp, speed int)' 127.0.0.1:7111/rest/sql system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d ' create table d1.t1 (ts timestamp, speed int)' 127.0.0.1:7111/rest/sql
print 22-> $system_content print 22-> $system_content
if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":0}@ then if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":1}@ then
return -1 return -1
endi endi
...@@ -214,13 +214,13 @@ endi ...@@ -214,13 +214,13 @@ endi
system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d 'create database d2' 127.0.0.1:7111/rest/sql system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d 'create database d2' 127.0.0.1:7111/rest/sql
print 28-> $system_content print 28-> $system_content
if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":0}@ then if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":1}@ then
return -1 return -1
endi endi
system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d ' create table d2.t1 (ts timestamp, speed int)' 127.0.0.1:7111/rest/sql system_content curl -H 'Authorization: Taosd /KfeAzX/f9na8qdtNZmtONryp201ma04bEl8LcvLUd7a8qdtNZmtONryp201ma04' -d ' create table d2.t1 (ts timestamp, speed int)' 127.0.0.1:7111/rest/sql
print 29-> $system_content print 29-> $system_content
if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":0}@ then if $system_content != @{"status":"succ","head":["affected_rows"],"data":[[0]],"rows":1}@ then
return -1 return -1
endi endi
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或