Skip to content

T
TDengine

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

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

more work

上级 81897d32
显示空白变更内容
内联 并排
Showing with 510 addition and 349 deletion
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 1753d121
...@@ -77,6 +77,9 @@ typedef struct STsdbFSState STsdbFSState; ...@@ -77,6 +77,9 @@ typedef struct STsdbFSState STsdbFSState;
#define VERSION_MIN 0 #define VERSION_MIN 0
#define VERSION_MAX INT64_MAX #define VERSION_MAX INT64_MAX
#define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN})
#define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX})
// tsdbUtil.c ============================================================================================== // tsdbUtil.c ==============================================================================================
// TSDBROW // TSDBROW
#define TSDBROW_SVERSION(ROW) TD_ROW_SVER((ROW)->pTSRow) #define TSDBROW_SVERSION(ROW) TD_ROW_SVER((ROW)->pTSRow)
...@@ -110,14 +113,15 @@ int32_t tGetKEYINFO(uint8_t *p, KEYINFO *pKeyInfo); ...@@ -110,14 +113,15 @@ int32_t tGetKEYINFO(uint8_t *p, KEYINFO *pKeyInfo);
int32_t tPutBlockCol(uint8_t *p, void *ph); int32_t tPutBlockCol(uint8_t *p, void *ph);
int32_t tGetBlockCol(uint8_t *p, void *ph); int32_t tGetBlockCol(uint8_t *p, void *ph);
// SBlock // SBlock
#define tBlockInit() ((SBlock){.info = tKEYINFOInit()}) #define tBlockInit() ((SBlock){0})
void tBlockReset(SBlock *pBlock); void tBlockReset(SBlock *pBlock);
void tBlockClear(SBlock *pBlock); void tBlockClear(SBlock *pBlock);
int32_t tPutBlock(uint8_t *p, void *ph); int32_t tPutBlock(uint8_t *p, void *ph);
int32_t tGetBlock(uint8_t *p, void *ph); int32_t tGetBlock(uint8_t *p, void *ph);
int32_t tBlockCmprFn(const void *p1, const void *p2); int32_t tBlockCmprFn(const void *p1, const void *p2);
// SBlockIdx // SBlockIdx
#define tBlockIdxInit(SUID, UID) ((SBlockIdx){.suid = (SUID), .uid = (UID), .info = tKEYINFOInit()}) // #define tBlockIdxInit(SUID, UID) ((SBlockIdx){.suid = (SUID), .uid = (UID), .info = tKEYINFOInit()})
void tBlockIdxReset(SBlockIdx *pBlockIdx);
int32_t tPutBlockIdx(uint8_t *p, void *ph); int32_t tPutBlockIdx(uint8_t *p, void *ph);
int32_t tGetBlockIdx(uint8_t *p, void *ph); int32_t tGetBlockIdx(uint8_t *p, void *ph);
// SColdata // SColdata
...@@ -328,7 +332,10 @@ struct TSDBROW { ...@@ -328,7 +332,10 @@ struct TSDBROW {
struct SBlockIdx { struct SBlockIdx {
int64_t suid; int64_t suid;
int64_t uid; int64_t uid;
KEYINFO info; TSKEY minKey;
TSKEY maxKey;
int64_t minVersion;
int64_t maxVersion;
int64_t offset; int64_t offset;
int64_t size; int64_t size;
}; };
...@@ -358,7 +365,10 @@ typedef struct { ...@@ -358,7 +365,10 @@ typedef struct {
} SSubBlock; } SSubBlock;
struct SBlock { struct SBlock {
KEYINFO info; TSDBKEY minKey;
TSDBKEY maxKey;
int64_t minVersion;
int64_t maxVersion;
int32_t nRow; int32_t nRow;
int8_t last; int8_t last;
int8_t hasDup; int8_t hasDup;
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 1753d121
...@@ -143,7 +143,7 @@ _err: ...@@ -143,7 +143,7 @@ _err:
static int32_t tsdbCommitTableDel(SCommitter *pCommitter, STbData *pTbData, SDelIdx *pDelIdx) { static int32_t tsdbCommitTableDel(SCommitter *pCommitter, STbData *pTbData, SDelIdx *pDelIdx) {
int32_t code = 0; int32_t code = 0;
SDelData *pDelData; SDelData *pDelData = &(SDelData){};
tb_uid_t suid; tb_uid_t suid;
tb_uid_t uid; tb_uid_t uid;
SDelIdx delIdx; // TODO SDelIdx delIdx; // TODO
...@@ -338,319 +338,320 @@ _err: ...@@ -338,319 +338,320 @@ _err:
#define ROW_END(pRow, maxKey) (((pRow) == NULL) || ((pRow)->pTSRow->ts > (maxKey))) #define ROW_END(pRow, maxKey) (((pRow) == NULL) || ((pRow)->pTSRow->ts > (maxKey)))
static int32_t tsdbCommitMemoryData(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, TSDBKEY eKey, // static int32_t tsdbCommitMemoryData(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, TSDBKEY eKey,
bool toDataOnly) { // bool toDataOnly) {
int32_t code = 0; // int32_t code = 0;
TSDBROW *pRow; // TSDBROW *pRow;
STSchema *pTSchema = NULL; // TODO // STSchema *pTSchema = NULL; // TODO
TSDBKEY key; // TSDBKEY key;
SBlock *pBlock = &pCommitter->nBlock; // SBlock *pBlock = &pCommitter->nBlock;
if (pIter == NULL) goto _exit; // if (pIter == NULL) goto _exit;
tBlockReset(pBlock); // tBlockReset(pBlock);
tBlockDataReset(&pCommitter->nBlockData); // tBlockDataReset(&pCommitter->nBlockData);
while (true) { // while (true) {
pRow = tsdbTbDataIterGet(pIter); // pRow = tsdbTbDataIterGet(pIter);
if (pRow == NULL || tsdbKeyCmprFn(&(TSDBKEY){.ts = pRow->pTSRow->ts, .version = pRow->version}, &eKey) > 0) { // if (pRow == NULL || tsdbKeyCmprFn(&(TSDBKEY){.ts = pRow->pTSRow->ts, .version = pRow->version}, &eKey) > 0) {
if (pCommitter->nBlockData.nRow == 0) { // if (pCommitter->nBlockData.nRow == 0) {
break; // break;
} else { // } else {
goto _write_block_data; // goto _write_block_data;
} // }
} // }
// update schema // // update schema
if (pTSchema == NULL || pTSchema->version != TSDBROW_SVERSION(pRow)) { // if (pTSchema == NULL || pTSchema->version != TSDBROW_SVERSION(pRow)) {
// TODO // // TODO
// pTSchema = NULL; // // pTSchema = NULL;
} // }
// append row // // append row
code = tBlockDataAppendRow(&pCommitter->nBlockData, pRow, pTSchema); // code = tBlockDataAppendRow(&pCommitter->nBlockData, pRow, pTSchema);
if (code) goto _err; // if (code) goto _err;
// update info // // update info
key = tsdbRowKey(pRow); // key = tsdbRowKey(pRow);
if (tsdbKeyCmprFn(&key, &pBlock->info.maxKey) > 0) pBlock->info.maxKey = key; // if (tsdbKeyCmprFn(&key, &pBlock->info.maxKey) > 0) pBlock->info.maxKey = key;
if (tsdbKeyCmprFn(&key, &pBlock->info.minKey) < 0) pBlock->info.minKey = key; // if (tsdbKeyCmprFn(&key, &pBlock->info.minKey) < 0) pBlock->info.minKey = key;
if (key.version > pBlock->info.maxVersion) pBlock->info.maxVersion = key.version; // if (key.version > pBlock->info.maxVersion) pBlock->info.maxVersion = key.version;
if (key.version < pBlock->info.minVerion) pBlock->info.minVerion = key.version; // if (key.version < pBlock->info.minVerion) pBlock->info.minVerion = key.version;
// iter next // // iter next
tsdbTbDataIterNext(pIter); // tsdbTbDataIterNext(pIter);
// check write // // check write
if (pCommitter->nBlockData.nRow < pCommitter->maxRow * 4 / 5) { // if (pCommitter->nBlockData.nRow < pCommitter->maxRow * 4 / 5) {
continue; // continue;
} // }
_write_block_data: // _write_block_data:
if (!toDataOnly && pCommitter->nBlockData.nRow < pCommitter->minKey) { // if (!toDataOnly && pCommitter->nBlockData.nRow < pCommitter->minKey) {
pCommitter->nBlock.last = 1; // pCommitter->nBlock.last = 1;
} else { // } else {
pCommitter->nBlock.last = 0; // pCommitter->nBlock.last = 0;
} // }
code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, pBlock); // code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, pBlock);
if (code) goto _err; // if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlockMap, pBlock, tPutBlock); // code = tMapDataPutItem(&pCommitter->nBlockMap, pBlock, tPutBlock);
if (code) goto _err; // if (code) goto _err;
// update info // // update info
if (tsdbKeyCmprFn(&pBlock->info.minKey, &pBlockIdx->info.minKey) < 0) pBlock->info.minKey = pBlockIdx->info.minKey; // if (tsdbKeyCmprFn(&pBlock->info.minKey, &pBlockIdx->info.minKey) < 0) pBlock->info.minKey =
if (tsdbKeyCmprFn(&pBlock->info.maxKey, &pBlockIdx->info.maxKey) < 0) pBlock->info.maxKey = pBlockIdx->info.maxKey; // pBlockIdx->info.minKey; if (tsdbKeyCmprFn(&pBlock->info.maxKey, &pBlockIdx->info.maxKey) < 0) pBlock->info.maxKey
if (pBlock->info.minVerion < pBlockIdx->info.minVerion) pBlockIdx->info.minVerion = pBlock->info.minVerion; // = pBlockIdx->info.maxKey; if (pBlock->info.minVerion < pBlockIdx->info.minVerion) pBlockIdx->info.minVerion =
if (pBlock->info.maxVersion < pBlockIdx->info.maxVersion) pBlockIdx->info.maxVersion = pBlock->info.maxVersion; // pBlock->info.minVerion; if (pBlock->info.maxVersion < pBlockIdx->info.maxVersion) pBlockIdx->info.maxVersion =
// pBlock->info.maxVersion;
tBlockReset(pBlock);
tBlockDataReset(&pCommitter->nBlockData); // tBlockReset(pBlock);
} // tBlockDataReset(&pCommitter->nBlockData);
// }
_exit:
return code; // _exit:
// return code;
_err:
tsdbError("vgId:%d commit memory data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code)); // _err:
return code; // tsdbError("vgId:%d commit memory data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
} // return code;
// }
static int32_t tsdbGetOverlapRowNumber(STbDataIter *pIter, SBlock *pBlock) {
int32_t nRow = 0; // static int32_t tsdbGetOverlapRowNumber(STbDataIter *pIter, SBlock *pBlock) {
TSDBROW *pRow; // int32_t nRow = 0;
TSDBKEY key; // TSDBROW *pRow;
int32_t c = 0; // TSDBKEY key;
STbDataIter iter = *pIter; // int32_t c = 0;
// STbDataIter iter = *pIter;
iter.pRow = NULL;
while (true) { // iter.pRow = NULL;
pRow = tsdbTbDataIterGet(pIter); // while (true) {
// pRow = tsdbTbDataIterGet(pIter);
if (pRow == NULL) break;
key = tsdbRowKey(pRow); // if (pRow == NULL) break;
// key = tsdbRowKey(pRow);
c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
if (c == 0) { // c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
nRow++; // if (c == 0) {
} else if (c > 0) { // nRow++;
break; // } else if (c > 0) {
} else { // break;
ASSERT(0); // } else {
} // ASSERT(0);
} // }
// }
return nRow;
} // return nRow;
// }
static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock,
int8_t toDataOnly) { // static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock,
int32_t code = 0; // int8_t toDataOnly) {
int32_t iRow = 0; // int32_t code = 0;
int32_t nRow = 0; // int32_t iRow = 0;
int32_t c; // int32_t nRow = 0;
TSDBROW *pRow; // int32_t c;
SBlock block = tBlockInit(); // TSDBROW *pRow;
TSDBKEY key1; // SBlock block = tBlockInit();
TSDBKEY key2; // TSDBKEY key1;
// TSDBKEY key2;
tBlockDataReset(&pCommitter->nBlockData);
// tBlockDataReset(&pCommitter->nBlockData);
// load last and merge until {pCommitter->maxKey, INT64_MAX}
code = tsdbReadBlockData(pCommitter->pReader, pBlockIdx, pBlock, &pCommitter->oBlockData, NULL, 0, NULL, NULL); // // load last and merge until {pCommitter->maxKey, INT64_MAX}
if (code) goto _err; // code = tsdbReadBlockData(pCommitter->pReader, pBlockIdx, pBlock, &pCommitter->oBlockData, NULL, 0, NULL, NULL);
// if (code) goto _err;
iRow = 0;
nRow = pCommitter->oBlockData.nRow; // iRow = 0;
pRow = tsdbTbDataIterGet(pIter); // nRow = pCommitter->oBlockData.nRow;
// pRow = tsdbTbDataIterGet(pIter);
while (true) {
if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && (iRow >= nRow)) { // while (true) {
if (pCommitter->nBlockData.nRow > 0) { // if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && (iRow >= nRow)) {
goto _write_block_data; // if (pCommitter->nBlockData.nRow > 0) {
} else { // goto _write_block_data;
break; // } else {
} // break;
} // }
// }
// TODO
// // TODO
_write_block_data:
block.last = pCommitter->nBlockData.nRow < pCommitter->minRow ? 1 : 0; // _write_block_data:
code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, &block); // block.last = pCommitter->nBlockData.nRow < pCommitter->minRow ? 1 : 0;
if (code) goto _err; // code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->nBlockData, NULL, NULL, pBlockIdx, &block);
// if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlockMap, &block, tPutBlock);
if (code) goto _err; // code = tMapDataPutItem(&pCommitter->nBlockMap, &block, tPutBlock);
} // if (code) goto _err;
// }
tBlockReset(&block);
tBlockDataReset(&pCommitter->nBlockData); // tBlockReset(&block);
// tBlockDataReset(&pCommitter->nBlockData);
return code;
// return code;
_err:
tsdbError("vgId:%d merge commit impl failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code)); // _err:
return code; // tsdbError("vgId:%d merge commit impl failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
} // return code;
// }
static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock,
int8_t isLastBlock) { // static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock,
int32_t code = 0; // int8_t isLastBlock) {
TSDBROW *pRow; // int32_t code = 0;
TSDBKEY key; // TSDBROW *pRow;
int32_t c; // TSDBKEY key;
// int32_t c;
if (pBlock == NULL) { // (pIter && pBlock == NULL)
key.ts = pCommitter->maxKey; // if (pBlock == NULL) { // (pIter && pBlock == NULL)
key.version = INT64_MAX; // key.ts = pCommitter->maxKey;
code = tsdbCommitMemoryData(pCommitter, pBlockIdx, pIter, key, 0); // key.version = INT64_MAX;
if (code) goto _err; // code = tsdbCommitMemoryData(pCommitter, pBlockIdx, pIter, key, 0);
} else if (pBlock->last) { // if (code) goto _err;
// merge // } else if (pBlock->last) {
code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 0); // // merge
if (code) goto _err; // code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 0);
} else { // pBlock && pBlock->last == 0 && (pIter == NULL || pIter) // if (code) goto _err;
// memory // } else { // pBlock && pBlock->last == 0 && (pIter == NULL || pIter)
if (pIter) { // // memory
key.ts = pBlock->info.minKey.ts; // if (pIter) {
key.version = pBlock->info.minKey.version - 1; // key.ts = pBlock->info.minKey.ts;
code = tsdbCommitMemoryData(pCommitter, pBlockIdx, pIter, key, 1); // key.version = pBlock->info.minKey.version - 1;
if (code) goto _err; // code = tsdbCommitMemoryData(pCommitter, pBlockIdx, pIter, key, 1);
} // if (code) goto _err;
// }
// merge or move block
pRow = tsdbTbDataIterGet(pIter); // // merge or move block
key.ts = pRow->pTSRow->ts; // pRow = tsdbTbDataIterGet(pIter);
key.version = pRow->version; // key.ts = pRow->pTSRow->ts;
// key.version = pRow->version;
c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
if (c > 0) { // c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
// move block // if (c > 0) {
code = tMapDataPutItem(&pCommitter->nBlockMap, pBlock, tPutBlock); // // move block
if (code) goto _err; // code = tMapDataPutItem(&pCommitter->nBlockMap, pBlock, tPutBlock);
} else if (c == 0) { // if (code) goto _err;
int32_t nOverlap = tsdbGetOverlapRowNumber(pIter, pBlock); // } else if (c == 0) {
// int32_t nOverlap = tsdbGetOverlapRowNumber(pIter, pBlock);
if (pBlock->nRow + nOverlap > pCommitter->maxRow || pBlock->nSubBlock == TSDB_MAX_SUBBLOCKS) {
code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 1); // if (pBlock->nRow + nOverlap > pCommitter->maxRow || pBlock->nSubBlock == TSDB_MAX_SUBBLOCKS) {
if (code) goto _err; // code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 1);
} else { // if (code) goto _err;
// add as a subblock // } else {
} // // add as a subblock
} else { // }
ASSERT(0); // } else {
} // ASSERT(0);
} // }
// }
return code;
// return code;
_err:
tsdbError("vgId:%d merge commit failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code)); // _err:
return code; // tsdbError("vgId:%d merge commit failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
} // return code;
// }
static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBlockIdx *pBlockIdx) {
int32_t code = 0; // static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBlockIdx *pBlockIdx) {
STbDataIter iter; // int32_t code = 0;
STbDataIter *pIter = &iter; // STbDataIter iter;
TSDBROW *pRow; // STbDataIter *pIter = &iter;
int64_t suid; // TSDBROW *pRow;
int64_t uid; // int64_t suid;
SBlockIdx blockIdx; // int64_t uid;
// SBlockIdx blockIdx;
// create iter
if (pTbData) { // // create iter
suid = pTbData->suid; // if (pTbData) {
uid = pTbData->uid; // suid = pTbData->suid;
tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = 0}, 0, pIter); // uid = pTbData->uid;
} else { // tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = 0}, 0, pIter);
suid = pBlockIdx->suid; // } else {
uid = pBlockIdx->uid; // suid = pBlockIdx->suid;
pIter = NULL; // uid = pBlockIdx->uid;
} // pIter = NULL;
// }
// check
pRow = tsdbTbDataIterGet(pIter); // // check
if (ROW_END(pRow, pCommitter->maxKey) && pBlockIdx == NULL) goto _exit; // pRow = tsdbTbDataIterGet(pIter);
// if (ROW_END(pRow, pCommitter->maxKey) && pBlockIdx == NULL) goto _exit;
// start ================================
tMapDataReset(&pCommitter->oBlockMap); // // start ================================
tBlockReset(&pCommitter->oBlock); // tMapDataReset(&pCommitter->oBlockMap);
tBlockDataReset(&pCommitter->oBlockData); // tBlockReset(&pCommitter->oBlock);
if (pBlockIdx) { // tBlockDataReset(&pCommitter->oBlockData);
code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlockMap, NULL); // if (pBlockIdx) {
if (code) goto _err; // code = tsdbReadBlock(pCommitter->pReader, pBlockIdx, &pCommitter->oBlockMap, NULL);
} // if (code) goto _err;
// }
blockIdx = tBlockIdxInit(suid, uid);
tMapDataReset(&pCommitter->nBlockMap); // blockIdx = tBlockIdxInit(suid, uid);
tBlockReset(&pCommitter->nBlock); // tMapDataReset(&pCommitter->nBlockMap);
tBlockDataReset(&pCommitter->nBlockData); // tBlockReset(&pCommitter->nBlock);
// tBlockDataReset(&pCommitter->nBlockData);
// impl ===============================
int32_t iBlock = 0; // // impl ===============================
int32_t nBlock = pCommitter->oBlockMap.nItem; // int32_t iBlock = 0;
// int32_t nBlock = pCommitter->oBlockMap.nItem;
// merge
pRow = tsdbTbDataIterGet(pIter); // // merge
while (!ROW_END(pRow, pCommitter->maxKey) && iBlock < nBlock) { // pRow = tsdbTbDataIterGet(pIter);
tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, &pCommitter->oBlock, tGetBlock); // while (!ROW_END(pRow, pCommitter->maxKey) && iBlock < nBlock) {
code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, &pCommitter->oBlock, iBlock == (nBlock - 1)); // tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, &pCommitter->oBlock, tGetBlock);
if (code) goto _err; // code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, &pCommitter->oBlock, iBlock == (nBlock - 1));
// if (code) goto _err;
pRow = tsdbTbDataIterGet(pIter);
iBlock++; // pRow = tsdbTbDataIterGet(pIter);
} // iBlock++;
// }
// mem
pRow = tsdbTbDataIterGet(pIter); // // mem
while (!ROW_END(pRow, pCommitter->maxKey)) { // pRow = tsdbTbDataIterGet(pIter);
code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, NULL, 0); // while (!ROW_END(pRow, pCommitter->maxKey)) {
if (code) goto _err; // code = tsdbMergeCommit(pCommitter, &blockIdx, pIter, NULL, 0);
// if (code) goto _err;
pRow = tsdbTbDataIterGet(pIter);
} // pRow = tsdbTbDataIterGet(pIter);
// }
// disk
while (iBlock < nBlock) { // // disk
tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, &pCommitter->oBlock, tGetBlock); // while (iBlock < nBlock) {
// tMapDataGetItemByIdx(&pCommitter->oBlockMap, iBlock, &pCommitter->oBlock, tGetBlock);
code = tsdbMergeCommit(pCommitter, &blockIdx, NULL, &pCommitter->oBlock, 0);
if (code) goto _err; // code = tsdbMergeCommit(pCommitter, &blockIdx, NULL, &pCommitter->oBlock, 0);
// if (code) goto _err;
iBlock++;
} // iBlock++;
// }
// end ===============================
code = tsdbWriteBlock(pCommitter->pWriter, &pCommitter->nBlockMap, NULL, &blockIdx); // // end ===============================
if (code) goto _err; // code = tsdbWriteBlock(pCommitter->pWriter, &pCommitter->nBlockMap, NULL, &blockIdx);
// if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlockIdxMap, &blockIdx, tPutBlockIdx);
if (code) goto _err; // code = tMapDataPutItem(&pCommitter->nBlockIdxMap, &blockIdx, tPutBlockIdx);
// if (code) goto _err;
_exit:
pRow = tsdbTbDataIterGet(pIter); // _exit:
if (pRow) { // pRow = tsdbTbDataIterGet(pIter);
ASSERT(pRow->pTSRow->ts > pCommitter->maxKey); // if (pRow) {
if (pCommitter->nextKey > pRow->pTSRow->ts) { // ASSERT(pRow->pTSRow->ts > pCommitter->maxKey);
pCommitter->nextKey = pRow->pTSRow->ts; // if (pCommitter->nextKey > pRow->pTSRow->ts) {
} // pCommitter->nextKey = pRow->pTSRow->ts;
} // }
// }
return code;
// return code;
_err:
tsdbError("vgId:%d commit Table data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code)); // _err:
return code; // tsdbError("vgId:%d commit Table data failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
} // return code;
// }
static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
STsdb *pTsdb = pCommitter->pTsdb; STsdb *pTsdb = pCommitter->pTsdb;
SDFileSet *pRSet = NULL; SDFileSet *pRSet = NULL;
SDFileSet *pWSet = NULL; SDFileSet wSet;
// memory // memory
pCommitter->nextKey = TSKEY_MAX; pCommitter->nextKey = TSKEY_MAX;
...@@ -675,22 +676,30 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) { ...@@ -675,22 +676,30 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
tBlockReset(&pCommitter->nBlock); tBlockReset(&pCommitter->nBlock);
tBlockDataReset(&pCommitter->nBlockData); tBlockDataReset(&pCommitter->nBlockData);
if (pRSet) { if (pRSet) {
pWSet = &(SDFileSet){.diskId = pRSet->diskId, wSet = (SDFileSet){.diskId = pRSet->diskId,
.fid = pCommitter->commitFid, .fid = pCommitter->commitFid,
.fHead = {.commitID = pCommitter->commitID, .offset = 0, .size = 0}, .fHead = {.commitID = pCommitter->commitID, .offset = 0, .size = 0},
.fData = pRSet->fData, .fData = pRSet->fData,
.fLast = {.commitID = pCommitter->commitID, .size = 0}, .fLast = {.commitID = pCommitter->commitID, .size = 0},
.fSma = pRSet->fSma}; .fSma = pRSet->fSma};
} else { } else {
SDiskID did = {.level = 0, .id = 0}; // TODO: alloc a new one STfs *pTfs = pTsdb->pVnode->pTfs;
pWSet = &(SDFileSet){.diskId = did, SDiskID did = {.level = 0, .id = 0};
// TODO: alloc a new disk
// tfsAllocDisk(pTfs, 0, &did);
// create the directory
tfsMkdirRecurAt(pTfs, pTsdb->path, did);
wSet = (SDFileSet){.diskId = did,
.fid = pCommitter->commitFid, .fid = pCommitter->commitFid,
.fHead = {.commitID = pCommitter->commitID, .offset = 0, .size = 0}, .fHead = {.commitID = pCommitter->commitID, .offset = 0, .size = 0},
.fData = {.commitID = pCommitter->commitID, .size = 0}, .fData = {.commitID = pCommitter->commitID, .size = 0},
.fLast = {.commitID = pCommitter->commitID, .size = 0}, .fLast = {.commitID = pCommitter->commitID, .size = 0},
.fSma = {.commitID = pCommitter->commitID, .size = 0}}; .fSma = {.commitID = pCommitter->commitID, .size = 0}};
} }
code = tsdbDataFWriterOpen(&pCommitter->pWriter, pTsdb, pWSet); code = tsdbDataFWriterOpen(&pCommitter->pWriter, pTsdb, &wSet);
if (code) goto _err; if (code) goto _err;
_exit: _exit:
...@@ -701,6 +710,50 @@ _err: ...@@ -701,6 +710,50 @@ _err:
return code; return code;
} }
static int32_t tsdbCommitMemoryData(SCommitter *pCommitter, STbData *pTbData) {
int32_t code = 0;
STsdb *pTsdb = pCommitter->pTsdb;
STbDataIter *pIter = &(STbDataIter){0};
TSDBKEY key = {.ts = pCommitter->minKey, .version = VERSION_MIN};
TSDBROW *pRow;
// create iter
tsdbTbDataIterOpen(pTbData, &key, 0, pIter);
pRow = tsdbTbDataIterGet(pIter);
if (pRow == NULL || tsdbRowKey(pRow).ts > pCommitter->maxKey) goto _exit;
// main loop
SBlockIdx *pBlockIdx = &(SBlockIdx){.suid = pTbData->suid, .uid = pTbData->uid};
SBlock *pBlock = &pCommitter->nBlock;
SBlockData *pBlockData = &pCommitter->nBlockData;
tBlockIdxReset(pBlockIdx);
tBlockReset(pBlock);
tBlockDataReset(pBlockData);
while (pRow != NULL && tsdbRowKey(pRow).ts <= pCommitter->maxKey) {
code = tBlockDataAppendRow(pBlockData, pRow, NULL);
if (code) goto _err;
tsdbTbDataIterNext(pIter);
pRow = tsdbTbDataIterGet(pIter);
if (pBlockData->nRow >= pCommitter->maxRow * 4 / 5) {
// write the block and do something
}
}
_exit:
if (pRow) {
pCommitter->nextKey = TMIN(pCommitter->nextKey, tsdbRowKey(pRow).ts);
}
return code;
_err:
tsdbError("vgId:%d tsdb commit memory data failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code;
}
static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) { static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
int32_t c; int32_t c;
...@@ -711,8 +764,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) { ...@@ -711,8 +764,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
int32_t iBlockIdx = 0; int32_t iBlockIdx = 0;
int32_t nBlockIdx = pCommitter->oBlockIdxMap.nItem; int32_t nBlockIdx = pCommitter->oBlockIdxMap.nItem;
STbData *pTbData; STbData *pTbData;
SBlockIdx blockIdx; SBlockIdx *pBlockIdx = &(SBlockIdx){0};
SBlockIdx *pBlockIdx = &blockIdx;
ASSERT(nTbData > 0); ASSERT(nTbData > 0);
...@@ -723,6 +775,81 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) { ...@@ -723,6 +775,81 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
pBlockIdx = NULL; pBlockIdx = NULL;
} }
// merge
while (pTbData && pBlockIdx) {
c = tTABLEIDCmprFn(pTbData, pBlockIdx);
if (c == 0) {
// merge commit
// code = tsdbMergeCommit(pCommitter, pTbData, pBlockIdx);
// if (code) goto _err;
iTbData++;
iBlockIdx++;
if (iTbData < nTbData) {
pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
} else {
pTbData = NULL;
}
if (iBlockIdx < nBlockIdx) {
tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
} else {
pBlockIdx = NULL;
}
} else if (c < 0) {
// commit memory data
// code = tsdbCommitMemoryData(pCommitter, pTbData);
// if (code) goto _err;
iTbData++;
if (iTbData < nTbData) {
pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
} else {
pTbData = NULL;
}
} else {
// commit disk data
// code = tsdbCommitDiskData(pCommitter, pBlockIdx);
// if (code) goto _err;
iBlockIdx++;
if (iBlockIdx < nBlockIdx) {
tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
} else {
pBlockIdx = NULL;
}
}
}
// disk
while (pBlockIdx) {
// commit disk data
// code = tsdbCommitDiskData(pCommitter, pBlockIdx);
// if (code) goto _err;
iBlockIdx++;
if (iBlockIdx < nBlockIdx) {
tMapDataGetItemByIdx(&pCommitter->oBlockIdxMap, iBlockIdx, pBlockIdx, tGetBlockIdx);
} else {
pBlockIdx = NULL;
}
}
// memory
while (pTbData) {
// commit memory data
code = tsdbCommitMemoryData(pCommitter, pTbData);
if (code) goto _err;
iTbData++;
if (iTbData < nTbData) {
pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
} else {
pTbData = NULL;
}
}
#if 0
while (true) { while (true) {
if (pTbData == NULL && pBlockIdx == NULL) break; if (pTbData == NULL && pBlockIdx == NULL) break;
...@@ -784,6 +911,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) { ...@@ -784,6 +911,7 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
} }
continue; continue;
} }
#endif
return code; return code;
......
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
浏览文件 @ 1753d121
...@@ -608,7 +608,7 @@ int32_t tsdbReadBlockSMA(SDataFReader *pReader, SBlockSMA *pBlkSMA) { ...@@ -608,7 +608,7 @@ int32_t tsdbReadBlockSMA(SDataFReader *pReader, SBlockSMA *pBlkSMA) {
// SDataFWriter ==================================================== // SDataFWriter ====================================================
struct SDataFWriter { struct SDataFWriter {
STsdb *pTsdb; STsdb *pTsdb;
SDFileSet *pSet; SDFileSet wSet;
TdFilePtr pHeadFD; TdFilePtr pHeadFD;
TdFilePtr pDataFD; TdFilePtr pDataFD;
TdFilePtr pLastFD; TdFilePtr pLastFD;
...@@ -630,9 +630,8 @@ int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pS ...@@ -630,9 +630,8 @@ int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pS
goto _err; goto _err;
} }
pWriter->pTsdb = pTsdb; pWriter->pTsdb = pTsdb;
pWriter->pSet = pSet; pWriter->wSet = *pSet;
pSet = &pWriter->wSet;
// create the directory if not there
// head // head
flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC; flag = TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
...@@ -809,10 +808,10 @@ int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter, uint8_t **ppBuf) { ...@@ -809,10 +808,10 @@ int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter, uint8_t **ppBuf) {
int64_t size = TSDB_FHDR_SIZE; int64_t size = TSDB_FHDR_SIZE;
int64_t n; int64_t n;
uint8_t *pBuf = NULL; uint8_t *pBuf = NULL;
SHeadFile *pHeadFile = &pWriter->pSet->fHead; SHeadFile *pHeadFile = &pWriter->wSet.fHead;
SDataFile *pDataFile = &pWriter->pSet->fData; SDataFile *pDataFile = &pWriter->wSet.fData;
SLastFile *pLastFile = &pWriter->pSet->fLast; SLastFile *pLastFile = &pWriter->wSet.fLast;
SSmaFile *pSmaFile = &pWriter->pSet->fSma; SSmaFile *pSmaFile = &pWriter->wSet.fSma;
// alloc // alloc
if (!ppBuf) ppBuf = &pBuf; if (!ppBuf) ppBuf = &pBuf;
...@@ -904,7 +903,7 @@ _err: ...@@ -904,7 +903,7 @@ _err:
int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SMapData *mBlockIdx, uint8_t **ppBuf) { int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SMapData *mBlockIdx, uint8_t **ppBuf) {
int32_t code = 0; int32_t code = 0;
int64_t size = 0; int64_t size = 0;
SHeadFile *pHeadFile = &pWriter->pSet->fHead; SHeadFile *pHeadFile = &pWriter->wSet.fHead;
int64_t n = 0; int64_t n = 0;
uint8_t *pBuf = NULL; uint8_t *pBuf = NULL;
...@@ -946,7 +945,7 @@ _err: ...@@ -946,7 +945,7 @@ _err:
int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, uint8_t **ppBuf, SBlockIdx *pBlockIdx) { int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *mBlock, uint8_t **ppBuf, SBlockIdx *pBlockIdx) {
int32_t code = 0; int32_t code = 0;
SHeadFile *pHeadFile = &pWriter->pSet->fHead; SHeadFile *pHeadFile = &pWriter->wSet.fHead;
uint8_t *pBuf = NULL; uint8_t *pBuf = NULL;
int64_t size; int64_t size;
int64_t n; int64_t n;
......
source/dnode/vnode/src/tsdb/tsdbUtil.c
浏览文件 @ 1753d121
...@@ -277,13 +277,25 @@ static FORCE_INLINE int32_t tGetTSDBKEY(uint8_t *p, TSDBKEY *pKey) { ...@@ -277,13 +277,25 @@ static FORCE_INLINE int32_t tGetTSDBKEY(uint8_t *p, TSDBKEY *pKey) {
} }
// SBlockIdx ====================================================== // SBlockIdx ======================================================
void tBlockIdxReset(SBlockIdx *pBlockIdx) {
pBlockIdx->minKey = TSKEY_MAX;
pBlockIdx->maxKey = TSKEY_MIN;
pBlockIdx->minVersion = VERSION_MAX;
pBlockIdx->maxVersion = VERSION_MIN;
pBlockIdx->offset = -1;
pBlockIdx->size = -1;
}
int32_t tPutBlockIdx(uint8_t *p, void *ph) { int32_t tPutBlockIdx(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SBlockIdx *pBlockIdx = (SBlockIdx *)ph; SBlockIdx *pBlockIdx = (SBlockIdx *)ph;
n += tPutI64(p ? p + n : p, pBlockIdx->suid); n += tPutI64(p ? p + n : p, pBlockIdx->suid);
n += tPutI64(p ? p + n : p, pBlockIdx->uid); n += tPutI64(p ? p + n : p, pBlockIdx->uid);
n += tPutKEYINFO(p ? p + n : p, &pBlockIdx->info); n += tPutI64(p ? p + n : p, pBlockIdx->minKey);
n += tPutI64(p ? p + n : p, pBlockIdx->maxKey);
n += tPutI64v(p ? p + n : p, pBlockIdx->minVersion);
n += tPutI64v(p ? p + n : p, pBlockIdx->maxVersion);
n += tPutI64v(p ? p + n : p, pBlockIdx->offset); n += tPutI64v(p ? p + n : p, pBlockIdx->offset);
n += tPutI64v(p ? p + n : p, pBlockIdx->size); n += tPutI64v(p ? p + n : p, pBlockIdx->size);
...@@ -296,7 +308,10 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) { ...@@ -296,7 +308,10 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) {
n += tGetI64(p + n, &pBlockIdx->suid); n += tGetI64(p + n, &pBlockIdx->suid);
n += tGetI64(p + n, &pBlockIdx->uid); n += tGetI64(p + n, &pBlockIdx->uid);
n += tGetKEYINFO(p + n, &pBlockIdx->info); n += tGetI64(p + n, &pBlockIdx->minKey);
n += tGetI64(p + n, &pBlockIdx->maxKey);
n += tGetI64v(p + n, &pBlockIdx->minVersion);
n += tGetI64v(p + n, &pBlockIdx->maxVersion);
n += tGetI64v(p + n, &pBlockIdx->offset); n += tGetI64v(p + n, &pBlockIdx->offset);
n += tGetI64v(p + n, &pBlockIdx->size); n += tGetI64v(p + n, &pBlockIdx->size);
...@@ -305,7 +320,10 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) { ...@@ -305,7 +320,10 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) {
// SBlock ====================================================== // SBlock ======================================================
void tBlockReset(SBlock *pBlock) { void tBlockReset(SBlock *pBlock) {
pBlock->info = tKEYINFOInit(); pBlock->minKey = TSDBKEY_MAX;
pBlock->maxKey = TSDBKEY_MIN;
pBlock->minVersion = VERSION_MAX;
pBlock->maxVersion = VERSION_MIN;
pBlock->nRow = 0; pBlock->nRow = 0;
pBlock->last = -1; pBlock->last = -1;
pBlock->cmprAlg = -1; pBlock->cmprAlg = -1;
...@@ -328,7 +346,10 @@ int32_t tPutBlock(uint8_t *p, void *ph) { ...@@ -328,7 +346,10 @@ int32_t tPutBlock(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SBlock *pBlock = (SBlock *)ph; SBlock *pBlock = (SBlock *)ph;
n += tPutKEYINFO(p ? p + n : p, &pBlock->info); n += tPutTSDBKEY(p ? p + n : p, &pBlock->minKey);
n += tPutTSDBKEY(p ? p + n : p, &pBlock->maxKey);
n += tPutI64v(p ? p + n : p, pBlock->minVersion);
n += tPutI64v(p ? p + n : p, pBlock->maxVersion);
n += tPutI32v(p ? p + n : p, pBlock->nRow); n += tPutI32v(p ? p + n : p, pBlock->nRow);
n += tPutI8(p ? p + n : p, pBlock->last); n += tPutI8(p ? p + n : p, pBlock->last);
n += tPutI8(p ? p + n : p, pBlock->hasDup); n += tPutI8(p ? p + n : p, pBlock->hasDup);
...@@ -348,7 +369,10 @@ int32_t tGetBlock(uint8_t *p, void *ph) { ...@@ -348,7 +369,10 @@ int32_t tGetBlock(uint8_t *p, void *ph) {
int32_t n = 0; int32_t n = 0;
SBlock *pBlock = (SBlock *)ph; SBlock *pBlock = (SBlock *)ph;
n += tGetKEYINFO(p + n, &pBlock->info); n += tGetTSDBKEY(p + n, &pBlock->minKey);
n += tGetTSDBKEY(p + n, &pBlock->maxKey);
n += tGetI64v(p + n, &pBlock->minVersion);
n += tGetI64v(p + n, &pBlock->maxVersion);
n += tGetI32v(p + n, &pBlock->nRow); n += tGetI32v(p + n, &pBlock->nRow);
n += tGetI8(p + n, &pBlock->last); n += tGetI8(p + n, &pBlock->last);
n += tGetI8(p + n, &pBlock->hasDup); n += tGetI8(p + n, &pBlock->hasDup);
...@@ -369,9 +393,9 @@ int32_t tBlockCmprFn(const void *p1, const void *p2) { ...@@ -369,9 +393,9 @@ int32_t tBlockCmprFn(const void *p1, const void *p2) {
SBlock *pBlock1 = (SBlock *)p1; SBlock *pBlock1 = (SBlock *)p1;
SBlock *pBlock2 = (SBlock *)p2; SBlock *pBlock2 = (SBlock *)p2;
if (tsdbKeyCmprFn(&pBlock1->info.maxKey, &pBlock2->info.minKey) < 0) { if (tsdbKeyCmprFn(&pBlock1->maxKey, &pBlock2->minKey) < 0) {
return -1; return -1;
} else if (tsdbKeyCmprFn(&pBlock1->info.minKey, &pBlock2->info.maxKey) > 0) { } else if (tsdbKeyCmprFn(&pBlock1->minKey, &pBlock2->maxKey) > 0) {
return 1; return 1;
} }
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或