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

more work

This commit is contained in:
Hongze Cheng 2022-06-13 06:05:05 +00:00
parent 03f9012a3d
commit 8f08aedcde
4 changed files with 120 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -158,6 +158,7 @@ int32_t tPutDelIdx(uint8_t *p, SDelIdx *pDelIdx);
int32_t tGetDelIdx(uint8_t *p, SDelIdx *pDelIdx); int32_t tGetDelIdx(uint8_t *p, SDelIdx *pDelIdx);
// SDelData // SDelData
int32_t tDelDataClear(SDelData *pDelData);
int32_t tDelDataPutItem(SDelData *pDelData, SDelDataItem *pItem); int32_t tDelDataPutItem(SDelData *pDelData, SDelDataItem *pItem);
int32_t tDelDataGetItemByIdx(SDelData *pDelData, SDelDataItem *pItem, int32_t idx); int32_t tDelDataGetItemByIdx(SDelData *pDelData, SDelDataItem *pItem, int32_t idx);
int32_t tDelDataGetItem(SDelData *pDelData, SDelDataItem *pItem, int64_t version); int32_t tDelDataGetItem(SDelData *pDelData, SDelDataItem *pItem, int64_t version);

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

@ -37,13 +37,14 @@ struct SCommitter {
SArray *aOBlockIdx; SArray *aOBlockIdx;
SArray *aBlockIdx; SArray *aBlockIdx;
// commit table data // commit table data
STbData *pTbData;
SBlockIdx *pBlockIdx; SBlockIdx *pBlockIdx;
/* commit del */ /* commit del */
SDelFReader *pDelFReader; SDelFReader *pDelFReader;
SDelFWriter *pDelFWriter; SDelFWriter *pDelFWriter;
SDelIdx delIdxOld; SDelIdx delIdxOld;
SDelIdx delIdxNew; SDelIdx delIdxNew;
STbData *pTbData;
SDelIdxItem *pDelIdxItem;
SDelData delDataOld; SDelData delDataOld;
SDelData delDataNew; SDelData delDataNew;
SDelIdxItem delIdxItem; SDelIdxItem delIdxItem;
@ -207,30 +208,59 @@ _err:
static int32_t tsdbCommitTableDel(SCommitter *pCommitter); static int32_t tsdbCommitTableDel(SCommitter *pCommitter);
static int32_t tsdbCommitDelImpl(SCommitter *pCommitter) { static int32_t tsdbCommitDelImpl(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
STsdb *pTsdb = pCommitter->pTsdb; STsdb *pTsdb = pCommitter->pTsdb;
SMemTable *pMemTable = pTsdb->imem; SMemTable *pMemTable = pTsdb->imem;
int32_t iTbData = 0; int32_t c;
int32_t nTbData = taosArrayGetSize(pMemTable->aTbData); int32_t iTbData = 0;
int32_t iDelIdx = 0; int32_t nTbData = taosArrayGetSize(pMemTable->aTbData);
int32_t nDelIdx = 0; // TODO int32_t iDelIdxItem = 0;
STbData *pTbData = NULL; int32_t nDelIdxItem = pCommitter->delIdxOld.offset.nOffset;
SDelIdx *pDelIdx = NULL; STbData *pTbData = NULL;
SDelIdx delIdx; SDelIdxItem *pDelIdxItem = NULL;
SDelIdxItem item;
// if (iTbData < nTbData) { while (iTbData < nTbData || iDelIdxItem < nDelIdxItem) {
// pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData); pTbData = NULL;
// } pDelIdxItem = NULL;
// if (iDelIdx < nDelIdx) { if (iTbData < nTbData) {
// // tIMapGet(); pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
// pDelIdx = &delIdx; }
// } if (iDelIdxItem < nDelIdxItem) {
tDelIdxGetItemByIdx(&pCommitter->delIdxOld, &item, iDelIdxItem);
while (iTbData < nTbData || iDelIdx < nDelIdx) { pDelIdxItem = &item;
if (pTbData && pDelIdx) {
} else {
} }
if (pTbData && pDelIdxItem) {
c = tTABLEIDCmprFn(pTbData, pDelIdxItem);
if (c == 0) {
iTbData++;
iDelIdxItem++;
} else if (c < 0) {
iTbData++;
pDelIdxItem = NULL;
} else {
iDelIdxItem++;
pTbData = NULL;
}
} else {
if (pTbData) {
iTbData++;
}
if (pDelIdxItem) {
iDelIdxItem++;
}
}
if (pTbData && pTbData->pHead == NULL) {
pTbData = NULL;
}
if (pTbData == NULL && pDelIdxItem == NULL) continue;
// do merge
pCommitter->pTbData = pTbData;
pCommitter->pDelIdxItem = pDelIdxItem;
code = tsdbCommitTableDel(pCommitter); code = tsdbCommitTableDel(pCommitter);
if (code) goto _err; if (code) goto _err;
} }
@ -542,17 +572,41 @@ static int32_t tsdbCommitTableDataEnd(SCommitter *pCommitter) {
} }
static int32_t tsdbCommitTableDelStart(SCommitter *pCommitter) { static int32_t tsdbCommitTableDelStart(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
tb_uid_t suid;
tb_uid_t uid;
if (pCommitter->pTbData) {
suid = pCommitter->pTbData->suid;
uid = pCommitter->pTbData->uid;
}
// load old // load old
pCommitter->delDataOld = (SDelData){0}; pCommitter->delDataOld = (SDelData){0};
if (0) { if (pCommitter->pDelIdxItem) {
code = tsdbReadDelData(pCommitter->pDelFReader, NULL /*TODO*/, &pCommitter->delDataOld, &pCommitter->pBuf4); suid = pCommitter->pDelIdxItem->suid;
uid = pCommitter->pDelIdxItem->uid;
code =
tsdbReadDelData(pCommitter->pDelFReader, pCommitter->pDelIdxItem, &pCommitter->delDataOld, &pCommitter->pBuf5);
if (code) goto _err; if (code) goto _err;
} }
// prepare new // prepare new
pCommitter->delDataNew = (SDelData){0}; pCommitter->delDataNew.suid = suid;
pCommitter->delDataNew.uid = uid;
pCommitter->delDataNew.offset.flag = 0;
pCommitter->delDataNew.offset.nOffset = 0;
pCommitter->delDataNew.nData = 0;
pCommitter->delIdxItem = (SDelIdxItem){
.suid = suid,
.uid = uid,
.minKey = TSKEY_MAX,
.maxKey = TSKEY_MIN,
.minVersion = INT64_MAX,
.maxVersion = INT64_MIN,
.offset = -1,
.size = -1,
};
return code; return code;
@ -562,14 +616,40 @@ _err:
} }
static int32_t tsdbCommitTableDelImpl(SCommitter *pCommitter) { static int32_t tsdbCommitTableDelImpl(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
SDelOp *pDelOp = NULL; SDelDataItem item;
for (; pDelOp; pDelOp = pDelOp->pNext) { // old
SDelDataItem item = {.version = pDelOp->version, .sKey = pDelOp->sKey, .eKey = pDelOp->eKey}; if (pCommitter->pDelIdxItem) {
for (int32_t iDelIdxItem = 0; iDelIdxItem < pCommitter->delDataOld.offset.nOffset; iDelIdxItem++) {
code = tDelDataGetItemByIdx(&pCommitter->delDataOld, &item, iDelIdxItem);
if (code) goto _err;
code = tDelDataPutItem(&pCommitter->delDataNew, &item); code = tDelDataPutItem(&pCommitter->delDataNew, &item);
if (code) goto _err; if (code) goto _err;
// update index
if (item.version < pCommitter->delIdxItem.minVersion) pCommitter->delIdxItem.minVersion = item.version;
if (item.version > pCommitter->delIdxItem.maxVersion) pCommitter->delIdxItem.maxVersion = item.version;
if (item.sKey < pCommitter->delIdxItem.minKey) pCommitter->delIdxItem.minKey = item.sKey;
if (item.eKey > pCommitter->delIdxItem.maxKey) pCommitter->delIdxItem.maxKey = item.eKey;
}
}
// new
if (pCommitter->pTbData) {
for (SDelOp *pDelOp = pCommitter->pTbData->pHead; pDelOp; pDelOp = pDelOp->pNext) {
item = (SDelDataItem){.version = pDelOp->version, .sKey = pDelOp->sKey, .eKey = pDelOp->eKey};
code = tDelDataPutItem(&pCommitter->delDataNew, &item);
if (code) goto _err;
// update index
if (item.version < pCommitter->delIdxItem.minVersion) pCommitter->delIdxItem.minVersion = item.version;
if (item.version > pCommitter->delIdxItem.maxVersion) pCommitter->delIdxItem.maxVersion = item.version;
if (item.sKey < pCommitter->delIdxItem.minKey) pCommitter->delIdxItem.minKey = item.sKey;
if (item.eKey > pCommitter->delIdxItem.maxKey) pCommitter->delIdxItem.maxKey = item.eKey;
}
} }
return code; return code;

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

@ -147,7 +147,6 @@ int32_t tsdbWriteDelData(SDelFWriter *pWriter, SDelData *pDelData, uint8_t **ppB
// prepare // prepare
pDelData->delimiter = TSDB_FILE_DLMT; pDelData->delimiter = TSDB_FILE_DLMT;
// todo
// alloc // alloc
if (!ppBuf) ppBuf = &pBuf; if (!ppBuf) ppBuf = &pBuf;

7
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -350,6 +350,13 @@ static FORCE_INLINE int32_t tGetDelDataItem(uint8_t *p, SDelDataItem *pItem) {
} }
// SDelData ====================================================== // SDelData ======================================================
int32_t tDelDataClear(SDelData *pDelData) {
int32_t code = 0;
tsdbFree(pDelData->offset.pOffset);
tsdbFree(pDelData->pData);
return code;
}
int32_t tDelDataPutItem(SDelData *pDelData, SDelDataItem *pItem) { int32_t tDelDataPutItem(SDelData *pDelData, SDelDataItem *pItem) {
int32_t code = 0; int32_t code = 0;
uint32_t offset = pDelData->nData; uint32_t offset = pDelData->nData;