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

more code

This commit is contained in:
Hongze Cheng 2022-12-27 16:59:52 +08:00
parent 3b3b8d4f9a
commit 1427f9218c
1 changed files with 122 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
source/dnode/vnode/src/tsdb/tsdbCompact.c
View File

@ -65,12 +65,13 @@ typedef struct {
SRBTree rtree; SRBTree rtree;
STsdbDataIter *pIter; STsdbDataIter *pIter;
SBlockData bData; SBlockData bData;
SSkmInfo tbSkm;
} STsdbCompactor; } STsdbCompactor;
#define TSDB_FLG_DEEP_COMPACT 0x1 #define TSDB_FLG_DEEP_COMPACT 0x1
// ITER ========================= // ITER =========================
static int32_t tsdbDataIterNext(STsdbDataIter *pIter); static int32_t tsdbDataIterNext(STsdbDataIter *pIter, TABLEID *pExcludeTableId);
static int32_t tsdbDataIterCmprFn(const SRBTreeNode *n1, const SRBTreeNode *n2) { static int32_t tsdbDataIterCmprFn(const SRBTreeNode *n1, const SRBTreeNode *n2) {
const STsdbDataIter *pIter1 = (STsdbDataIter *)((char *)n1 - offsetof(STsdbDataIter, n)); const STsdbDataIter *pIter1 = (STsdbDataIter *)((char *)n1 - offsetof(STsdbDataIter, n));
@ -132,7 +133,7 @@ static int32_t tsdbDataDIterOpen(SDataFReader *pReader, STsdbDataIter **ppIter)
pDataDIter->iDataBlk = 0; pDataDIter->iDataBlk = 0;
pDataDIter->iRow = 0; pDataDIter->iRow = 0;
code = tsdbDataIterNext(pIter); code = tsdbDataIterNext(pIter, NULL);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
_exit: _exit:
@ -182,7 +183,7 @@ static int32_t tsdbSttDIterOpen(SDataFReader *pReader, int32_t iStt, STsdbDataIt
pSttDIter->iSttBlk = -1; pSttDIter->iSttBlk = -1;
pSttDIter->iRow = -1; pSttDIter->iRow = -1;
code = tsdbDataIterNext(pIter); code = tsdbDataIterNext(pIter, NULL);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
_exit: _exit:
@ -205,7 +206,7 @@ static void tsdbDataIterClose(STsdbDataIter *pIter) {
ASSERT(0); ASSERT(0);
} }
static int32_t tsdbDataIterNext(STsdbDataIter *pIter) { static int32_t tsdbDataIterNext(STsdbDataIter *pIter, TABLEID *pExcludeTableId) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
@ -218,25 +219,49 @@ static int32_t tsdbDataIterNext(STsdbDataIter *pIter) {
} else if (pIter->flag & TSDB_ITER_TYPE_STT) { } else if (pIter->flag & TSDB_ITER_TYPE_STT) {
SSttDIter *pSttDIter = (SSttDIter *)pIter->handle; SSttDIter *pSttDIter = (SSttDIter *)pIter->handle;
pSttDIter->iRow++; for (;;) {
if (pSttDIter->iRow < pSttDIter->bData.nRow) { if (++pSttDIter->iRow >= pSttDIter->bData.nRow) {
for (;;) {
if (++pSttDIter->iSttBlk < taosArrayGetSize(pSttDIter->aSttBlk)) {
SSttBlk *pSttBlk = (SSttBlk *)taosArrayGet(pSttDIter->aSttBlk, pSttDIter->iSttBlk);
// check exclusion
if (pExcludeTableId) {
if (pExcludeTableId->uid) { // exclude (suid, uid)
if (pSttBlk->minUid == pExcludeTableId->uid && pSttBlk->maxUid == pExcludeTableId->uid) continue;
} else { // exclude (suid, *)
if (pSttBlk->suid == pExcludeTableId->suid) continue;
}
}
code = tsdbReadSttBlockEx(pSttDIter->pReader, pSttDIter->iStt, pSttBlk, &pSttDIter->bData);
TSDB_CHECK_CODE(code, lino, _exit);
pIter->rowInfo.suid = pSttBlk->suid;
pSttDIter->iRow = 0;
break;
} else {
// iter end, all set 0 and exit
pIter->rowInfo.suid = 0;
pIter->rowInfo.uid = 0;
goto _exit;
}
}
}
pIter->rowInfo.uid = pSttDIter->bData.uid ? pSttDIter->bData.uid : pSttDIter->bData.aUid[pSttDIter->iRow]; pIter->rowInfo.uid = pSttDIter->bData.uid ? pSttDIter->bData.uid : pSttDIter->bData.aUid[pSttDIter->iRow];
pIter->rowInfo.row = tsdbRowFromBlockData(&pSttDIter->bData, pSttDIter->iRow); pIter->rowInfo.row = tsdbRowFromBlockData(&pSttDIter->bData, pSttDIter->iRow);
} else {
pSttDIter->iSttBlk++;
if (pSttDIter->iSttBlk < taosArrayGetSize(pSttDIter->aSttBlk)) {
code = tsdbReadSttBlockEx(pSttDIter->pReader, pSttDIter->iStt,
taosArrayGet(pSttDIter->aSttBlk, pSttDIter->iSttBlk), &pSttDIter->bData);
TSDB_CHECK_CODE(code, lino, _exit);
pSttDIter->iRow = 0; // check exclusion
pIter->rowInfo.suid = pSttDIter->bData.suid; if (pExcludeTableId) {
pIter->rowInfo.uid = pSttDIter->bData.uid ? pSttDIter->bData.uid : pSttDIter->bData.aUid[pSttDIter->iRow]; if (pExcludeTableId->uid) { // exclude (suid, uid)
pIter->rowInfo.row = tsdbRowFromBlockData(&pSttDIter->bData, pSttDIter->iRow); if (pIter->rowInfo.uid == pExcludeTableId->uid) continue;
} else { } else { // exclude (suid, *)
pIter->rowInfo.suid = 0; if (pIter->rowInfo.suid == pExcludeTableId->suid) continue;
pIter->rowInfo.uid = 0; }
} }
break;
} }
} else { } else {
ASSERT(0); ASSERT(0);
@ -330,37 +355,94 @@ _exit:
return code; return code;
} }
static int32_t tsdbCompactNextRowImpl(STsdbCompactor *pCompactor, TABLEID *pExcludeTableId) {
int32_t code = 0;
int32_t lino = 0;
for (;;) {
if (pCompactor->pIter) {
code = tsdbDataIterNext(pCompactor->pIter, pExcludeTableId);
TSDB_CHECK_CODE(code, lino, _exit);
if (pCompactor->pIter->rowInfo.suid == 0 && pCompactor->pIter->rowInfo.uid == 0) {
pCompactor->pIter = NULL;
} else {
SRBTreeNode *pNode = tRBTreeMin(&pCompactor->rtree);
if (pNode) {
STsdbDataIter *pIter = TSDB_DATA_ITER_FROM_RBTN(pNode);
int32_t c = tRowInfoCmprFn(&pCompactor->pIter->rowInfo, &pIter->rowInfo);
ASSERT(c);
if (c > 0) {
tRBTreePut(&pCompactor->rtree, &pCompactor->pIter->n);
pCompactor->pIter = NULL;
}
}
}
}
if (pCompactor->pIter == NULL) {
SRBTreeNode *pNode = tRBTreeMin(&pCompactor->rtree);
if (pNode) {
pCompactor->pIter = TSDB_DATA_ITER_FROM_RBTN(pNode);
tRBTreeDrop(&pCompactor->rtree, pNode);
if (pExcludeTableId) {
if (pExcludeTableId->uid) {
if (pCompactor->pIter->rowInfo.uid == pExcludeTableId->uid) continue;
} else {
if (pCompactor->pIter->rowInfo.suid == pExcludeTableId->suid) continue;
}
}
}
}
break;
}
_exit:
return code;
}
static int32_t tsdbCompactNextRow(STsdbCompactor *pCompactor) { static int32_t tsdbCompactNextRow(STsdbCompactor *pCompactor) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0; int32_t lino = 0;
if (pCompactor->pIter) { for (;;) {
code = tsdbDataIterNext(pCompactor->pIter); code = tsdbCompactNextRowImpl(pCompactor, NULL);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
if (pCompactor->pIter->rowInfo.suid == 0 && pCompactor->pIter->rowInfo.uid == 0) { // check if the table of the row exists
pCompactor->pIter = NULL; if (pCompactor->pIter) {
} else { if (pCompactor->pIter->rowInfo.suid == pCompactor->tbSkm.suid &&
SRBTreeNode *pNode = tRBTreeMin(&pCompactor->rtree); pCompactor->pIter->rowInfo.uid == pCompactor->tbSkm.uid) {
if (pNode) { break;
STsdbDataIter *pIter = TSDB_DATA_ITER_FROM_RBTN(pNode); } else {
SMetaInfo info;
if (metaGetInfo(pCompactor->pTsdb->pVnode->pMeta, pCompactor->pIter->rowInfo.uid, &info, NULL) !=
TSDB_CODE_SUCCESS) {
// table not exist
} else {
// update table schema
STSchema *pTSchema =
metaGetTbTSchema(pCompactor->pTsdb->pVnode->pMeta, pCompactor->pIter->rowInfo.uid, info.version, 1);
if (pTSchema == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
int32_t c = tRowInfoCmprFn(&pCompactor->pIter->rowInfo, &pIter->rowInfo); pCompactor->tbSkm.suid = pCompactor->pIter->rowInfo.suid;
ASSERT(c); pCompactor->tbSkm.uid = pCompactor->pIter->rowInfo.uid;
tDestroyTSchema(pCompactor->tbSkm.pTSchema);
pCompactor->tbSkm.pTSchema = pTSchema;
if (c > 0) { break;
tRBTreePut(&pCompactor->rtree, &pCompactor->pIter->n);
pCompactor->pIter = NULL;
} }
} }
} } else {
} // iter end, just break out
break;
if (pCompactor->pIter == NULL) {
SRBTreeNode *pNode = tRBTreeMin(&pCompactor->rtree);
if (pNode) {
pCompactor->pIter = TSDB_DATA_ITER_FROM_RBTN(pNode);
tRBTreeDrop(&pCompactor->rtree, pNode);
} }
} }