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 07:02:15 +00:00
parent 2c63430804
commit 627dc00b93
3 changed files with 174 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -112,14 +112,15 @@ typedef struct SCacheFReader SCacheFReader;
// tsdbCommit.c ==============================================================================================
// tsdbReadImpl.c ==============================================================================================
typedef struct SBlockIdx SBlockIdx;
typedef struct SBlockInfo SBlockInfo;
typedef struct SBlock SBlock;
typedef struct SBlockCol SBlockCol;
typedef struct SBlockStatis SBlockStatis;
typedef struct SAggrBlkCol SAggrBlkCol;
typedef struct SBlockData SBlockData;
typedef struct SReadH SReadH;
typedef struct SBlockIdxItem SBlockIdxItem;
typedef struct SBlockIdx SBlockIdx;
typedef struct SBlockInfo SBlockInfo;
typedef struct SBlock SBlock;
typedef struct SBlockCol SBlockCol;
typedef struct SBlockStatis SBlockStatis;
typedef struct SAggrBlkCol SAggrBlkCol;
typedef struct SBlockData SBlockData;
typedef struct SReadH SReadH;
typedef struct SDFileSetReader SDFileSetReader;
typedef struct SDFileSetWriter SDFileSetWriter;
@ -149,6 +150,16 @@ void tsdbFree(uint8_t *pBuf);
int32_t tTABLEIDCmprFn(const void *p1, const void *p2);
int32_t tsdbKeyCmprFn(const void *p1, const void *p2);
// SBlockIdx
int32_t tBlockIdxClear(SBlockIdx *pBlockIdx);
int32_t tBlockIdxPutItem(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem);
int32_t tBlockIdxGetItemByIdx(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem, int32_t idx);
int32_t tBlockIdxGetItem(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem, TABLEID id);
int32_t tPutBlockIdx(uint8_t *p, SBlockIdx *pBlockIdx);
int32_t tGetBlockIdx(uint8_t *p, SBlockIdx *pBlockIdx);
// SBlock
// SDelIdx
int32_t tDelIdxClear(SDelIdx *pDelIdx);
int32_t tDelIdxPutItem(SDelIdx *pDelIdx, SDelIdxItem *pItem);
@ -251,15 +262,24 @@ struct TSDBROW {
STSRow *pTSRow;
};
struct SBlockIdx {
struct SBlockIdxItem {
int64_t suid;
int64_t uid;
int64_t maxVersion;
TSDBKEY minKey;
TSDBKEY maxKey;
int64_t minVersion;
int64_t maxVersion;
int64_t offset;
int64_t size;
};
struct SBlockIdx {
uint32_t delimiter;
SOffset offset;
uint32_t nData;
uint8_t *pData;
};
typedef enum {
TSDB_SBLK_VER_0 = 0,
TSDB_SBLK_VER_MAX,
@ -267,19 +287,20 @@ typedef enum {
#define SBlockVerLatest TSDB_SBLK_VER_0
struct SBlock {
struct SBlockItem {
TSDBKEY minKey;
TSDBKEY maxKey;
int64_t minVerion;
int64_t maxVersion;
int64_t minVersion;
uint8_t flags; // last, algorithm
};
struct SBlockInfo {
int32_t delimiter; // For recovery usage
uint64_t suid;
uint64_t uid;
SBlock blocks[];
struct SBlock {
uint32_t delimiter;
int64_t suid;
int64_t uid;
SOffset offset;
uint32_t nData;
uint8_t *pData;
};
struct SBlockCol {

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

@ -295,6 +295,7 @@ static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) {
if (code) goto _err;
}
tDelDataClear(&pCommitter->delDataNew);
tDelIdxClear(&pCommitter->delIdxNew);
return code;

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

@ -203,6 +203,25 @@ int32_t tsdbKeyCmprFn(const void *p1, const void *p2) {
return 0;
}
// TSDBKEY ======================================================
static FORCE_INLINE int32_t tPutTSDBKEY(uint8_t *p, TSDBKEY *pKey) {
int32_t n = 0;
n += tPutI64v(p ? p + n : p, pKey->version);
n += tPutI64(p ? p + n : p, pKey->ts);
return n;
}
static FORCE_INLINE int32_t tGetTSDBKEY(uint8_t *p, TSDBKEY *pKey) {
int32_t n = 0;
n += tGetI64v(p + n, &pKey->version);
n += tGetI64(p + n, &pKey->ts);
return n;
}
// SDelIdxItem ======================================================
static FORCE_INLINE int32_t tPutDelIdxItem(uint8_t *p, SDelIdxItem *pDelIdxItem) {
int32_t n = 0;
@ -234,6 +253,116 @@ static FORCE_INLINE int32_t tGetDelIdxItem(uint8_t *p, SDelIdxItem *pDelIdxItem)
return n;
}
// SBlockIdxItem ======================================================
static FORCE_INLINE int32_t tPutBlockIdxItem(uint8_t *p, SBlockIdxItem *pItem) {
int32_t n = 0;
n += tPutI64(p ? p + n : p, pItem->suid);
n += tPutI64(p ? p + n : p, pItem->uid);
n += tPutTSDBKEY(p ? p + n : p, &pItem->minKey);
n += tPutTSDBKEY(p ? p + n : p, &pItem->maxKey);
n += tPutI64v(p ? p + n : p, pItem->minVersion);
n += tPutI64v(p ? p + n : p, pItem->maxVersion);
n += tPutI64v(p ? p + n : p, pItem->offset);
n += tPutI64v(p ? p + n : p, pItem->size);
return n;
}
static FORCE_INLINE int32_t tGetBlockIdxItem(uint8_t *p, SBlockIdxItem *pItem) {
int32_t n = 0;
n += tGetI64(p + n, &pItem->suid);
n += tGetI64(p + n, &pItem->uid);
n += tGetTSDBKEY(p + n, &pItem->minKey);
n += tGetTSDBKEY(p + n, &pItem->maxKey);
n += tGetI64v(p + n, &pItem->minVersion);
n += tGetI64v(p + n, &pItem->maxVersion);
n += tGetI64v(p + n, &pItem->offset);
n += tGetI64v(p + n, &pItem->size);
return n;
}
// SBlockIdx ======================================================
int32_t tBlockIdxClear(SBlockIdx *pBlockIdx) {
int32_t code = 0;
tsdbFree(pBlockIdx->offset.pOffset);
tsdbFree(pBlockIdx->pData);
return code;
}
int32_t tBlockIdxPutItem(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem) {
int32_t code = 0;
// TODO
return code;
}
int32_t tBlockIdxGetItemByIdx(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem, int32_t idx) {
int32_t code = 0;
int32_t offset;
offset = tsdbGetOffset(&pBlockIdx->offset, idx);
if (offset < 0) {
code = TSDB_CODE_NOT_FOUND;
goto _exit;
}
tGetBlockIdxItem(pBlockIdx->pData + offset, pItem);
_exit:
return code;
}
int32_t tBlockIdxGetItem(SBlockIdx *pBlockIdx, SBlockIdxItem *pItem, TABLEID id) {
int32_t code = 0;
int32_t lidx = 0;
int32_t ridx = pBlockIdx->offset.nOffset - 1;
int32_t midx;
int32_t c;
while (lidx <= ridx) {
midx = (lidx + midx) / 2;
code = tBlockIdxGetItemByIdx(pBlockIdx, pItem, midx);
if (code) goto _exit;
c = tTABLEIDCmprFn(&id, pItem);
if (c == 0) {
goto _exit;
} else if (c < 0) {
ridx = midx - 1;
} else {
lidx = midx + 1;
}
}
code = TSDB_CODE_NOT_FOUND;
_exit:
return code;
}
int32_t tPutBlockIdx(uint8_t *p, SBlockIdx *pBlockIdx) {
int32_t n = 0;
n += tPutU32(p ? p + n : p, pBlockIdx->delimiter);
n += tPutOffset(p ? p + n : p, &pBlockIdx->offset);
n += tPutBinary(p ? p + n : p, pBlockIdx->pData, pBlockIdx->nData);
return n;
}
int32_t tGetBlockIdx(uint8_t *p, SBlockIdx *pBlockIdx) {
int32_t n = 0;
n += tGetU32(p + n, &pBlockIdx->delimiter);
n += tGetOffset(p + n, &pBlockIdx->offset);
n += tGetBinary(p + n, &pBlockIdx->pData, &pBlockIdx->nData);
return n;
}
// SDelIdx ======================================================
int32_t tDelIdxClear(SDelIdx *pDelIdx) {
int32_t code = 0;
@ -279,12 +408,11 @@ _exit:
}
int32_t tDelIdxGetItem(SDelIdx *pDelIdx, SDelIdxItem *pItem, TABLEID id) {
int32_t code = 0;
int32_t lidx = 0;
int32_t ridx = pDelIdx->offset.nOffset - 1;
int32_t midx;
uint64_t offset;
int32_t c;
int32_t code = 0;
int32_t lidx = 0;
int32_t ridx = pDelIdx->offset.nOffset - 1;
int32_t midx;
int32_t c;
while (lidx <= ridx) {
midx = (lidx + ridx) / 2;