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-10 11:50:06 +00:00
parent 2f87d80d1c
commit 39626c1bc6
5 changed files with 109 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -78,6 +78,12 @@ int32_t tsdbFSEnd(STsdbFS *pFS, int8_t rollback);
typedef struct SDelData SDelData; typedef struct SDelData SDelData;
typedef struct SDelIdx SDelIdx; typedef struct SDelIdx SDelIdx;
// SDataFWriter
typedef struct SDataFWriter SDataFWriter;
// SDataFReader
typedef struct SDataFReader SDataFReader;
// SDelFWriter // SDelFWriter
typedef struct SDelFWriter SDelFWriter; typedef struct SDelFWriter SDelFWriter;
@ -89,11 +95,17 @@ int32_t tsdbWriteDelIdx(SDelFWriter *pWriter, SDelIdx *pDelIdx, uint8_t **ppBuf)
// SDelFReader // SDelFReader
typedef struct SDelFReader SDelFReader; typedef struct SDelFReader SDelFReader;
int32_t tsdbDelFReaderOpen(SDelFReader *pReader, SDelFile *pFile); int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile);
int32_t tsdbDelFReaderClose(SDelFReader *pReader); int32_t tsdbDelFReaderClose(SDelFReader *pReader);
int32_t tsdbReadDelData(SDelFReader *pReader, SDelData *pDelData, uint8_t **ppBuf); int32_t tsdbReadDelData(SDelFReader *pReader, SDelData *pDelData, uint8_t **ppBuf);
int32_t tsdbReadDelIdx(SDelFReader *pReader, SDelIdx *pDelIdx, uint8_t **ppBuf); int32_t tsdbReadDelIdx(SDelFReader *pReader, SDelIdx *pDelIdx, uint8_t **ppBuf);
// SCacheFWriter
typedef struct SCacheFWriter SCacheFWriter;
// SCacheFReader
typedef struct SCacheFReader SCacheFReader;
// tsdbCommit.c ============================================================================================== // tsdbCommit.c ==============================================================================================
// old // old
@ -314,7 +326,7 @@ struct SMemTable {
TSDBKEY minKey; TSDBKEY minKey;
TSDBKEY maxKey; TSDBKEY maxKey;
int64_t nRow; int64_t nRow;
int64_t nDelOp; int64_t nDel;
SArray *aTbData; // SArray<STbData*> SArray *aTbData; // SArray<STbData*>
}; };
@ -782,10 +794,11 @@ typedef struct {
struct SDelIdx { struct SDelIdx {
uint32_t delimiter; uint32_t delimiter;
int8_t flags; uint8_t flags;
int64_t nItem; uint32_t nOffset;
uint8_t *pOffset; uint8_t *pOffset;
uint8_t *pDelIdxItem; uint32_t nData;
uint8_t *pData;
}; };
#endif #endif

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

@ -21,6 +21,7 @@ struct SCommitter {
STsdb *pTsdb; STsdb *pTsdb;
uint8_t *pBuf1; uint8_t *pBuf1;
uint8_t *pBuf2; uint8_t *pBuf2;
uint8_t *pBuf3;
/* commit data */ /* commit data */
int32_t minutes; int32_t minutes;
int8_t precision; int8_t precision;
@ -37,12 +38,10 @@ struct SCommitter {
STbData *pTbData; STbData *pTbData;
SBlockIdx *pBlockIdx; SBlockIdx *pBlockIdx;
/* commit del */ /* commit del */
SDelFReader *pTombstoneReader; SDelFReader *pDelFReader;
SDelFWriter *pTombstoneWritter; SDelFWriter *pDelFWriter;
SDelIdx delIdxO; SDelIdx oDelIdx;
SDelIdx delIdxN; SDelIdx nDelIdx;
SDelData delDataO;
SDelData delDataN;
/* commit cache */ /* commit cache */
}; };
@ -164,8 +163,35 @@ _err:
} }
static int32_t tsdbCommitDelStart(SCommitter *pCommitter) { static int32_t tsdbCommitDelStart(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
// TODO STsdb *pTsdb = pCommitter->pTsdb;
SMemTable *pMemTable = pTsdb->imem;
SDelFile *pDelFileR = NULL; // TODO
SDelFile *pDelFileW = NULL; // TODO
if (pDelFileR) {
code = tsdbDelFReaderOpen(&pCommitter->pDelFReader, pDelFileR);
if (code) {
goto _err;
}
code = tsdbReadDelIdx(pCommitter->pDelFReader, &pCommitter->oDelIdx, &pCommitter->pBuf1);
if (code) {
goto _err;
}
}
code = tsdbDelFWriterOpen(&pCommitter->pDelFWriter, pDelFileW);
if (code) {
goto _err;
}
_exit:
tsdbDebug("vgId:%d commit del start", TD_VID(pTsdb->pVnode));
return code;
_err:
tsdbError("vgId:%d commit del start failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
@ -182,36 +208,36 @@ static int32_t tsdbCommitDelImpl(SCommitter *pCommitter) {
SDelIdx *pDelIdx = NULL; SDelIdx *pDelIdx = NULL;
while (iTbData < nTbData || iDelIdx < nDelIdx) { while (iTbData < nTbData || iDelIdx < nDelIdx) {
if (iTbData < nTbData) { // if (iTbData < nTbData) {
pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData); // pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
} else { // } else {
pTbData = NULL; // pTbData = NULL;
} // }
if (iDelIdx < nDelIdx) { // if (iDelIdx < nDelIdx) {
// pDelIdx = ; // TODO // // pDelIdx = ; // TODO
} else { // } else {
pDelIdx = NULL; // pDelIdx = NULL;
} // }
if (pTbData && pDelIdx) { // if (pTbData && pDelIdx) {
c = tTABLEIDCmprFn(pTbData, pDelIdx); // c = tTABLEIDCmprFn(pTbData, pDelIdx);
if (c == 0) { // if (c == 0) {
iTbData++; // iTbData++;
iDelIdx++; // iDelIdx++;
} else if (c < 0) { // } else if (c < 0) {
iTbData++; // iTbData++;
pDelIdx = NULL; // pDelIdx = NULL;
} else { // } else {
iDelIdx++; // iDelIdx++;
pTbData = NULL; // pTbData = NULL;
} // }
} else { // } else {
if (pTbData) { // if (pTbData) {
iTbData++; // iTbData++;
} else { // } else {
iDelIdx++; // iDelIdx++;
} // }
} // }
// TODO: commit with the pTbData and pDelIdx // TODO: commit with the pTbData and pDelIdx
} }
@ -222,8 +248,6 @@ static int32_t tsdbCommitDelImpl(SCommitter *pCommitter) {
static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) { static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) {
int32_t code = 0; int32_t code = 0;
ASSERT(pCommitter->delIdxN.nItem > 0);
return code; return code;
_err: _err:
@ -235,7 +259,7 @@ static int32_t tsdbCommitDel(SCommitter *pCommitter) {
STsdb *pTsdb = pCommitter->pTsdb; STsdb *pTsdb = pCommitter->pTsdb;
SMemTable *pMemTable = pTsdb->imem; SMemTable *pMemTable = pTsdb->imem;
if (pMemTable->nDelOp == 0) { if (pMemTable->nDel == 0) {
goto _exit; goto _exit;
} }
@ -258,9 +282,11 @@ static int32_t tsdbCommitDel(SCommitter *pCommitter) {
} }
_exit: _exit:
tsdbDebug("vgId:%d commit del data, nDel:%" PRId64, TD_VID(pTsdb->pVnode), pMemTable->nDel);
return code; return code;
_err: _err:
tsdbError("vgId:%d failed to commit del data since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
return code; return code;
} }

4
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -47,7 +47,7 @@ int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
pMemTable->minKey = (TSDBKEY){.ts = TSKEY_MAX, .version = INT64_MAX}; pMemTable->minKey = (TSDBKEY){.ts = TSKEY_MAX, .version = INT64_MAX};
pMemTable->maxKey = (TSDBKEY){.ts = TSKEY_MIN, .version = -1}; pMemTable->maxKey = (TSDBKEY){.ts = TSKEY_MIN, .version = -1};
pMemTable->nRow = 0; pMemTable->nRow = 0;
pMemTable->nDelOp = 0; pMemTable->nDel = 0;
pMemTable->aTbData = taosArrayInit(128, sizeof(STbData *)); pMemTable->aTbData = taosArrayInit(128, sizeof(STbData *));
if (pMemTable->aTbData == NULL) { if (pMemTable->aTbData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
@ -174,7 +174,7 @@ int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid
// update the state of pMemTable and other (todo) // update the state of pMemTable and other (todo)
pMemTable->nDelOp++; pMemTable->nDel++;
tsdbError("vgId:%d delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64 tsdbError("vgId:%d delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
" since %s", " since %s",

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

@ -109,7 +109,7 @@ struct SDelFReader {
TdFilePtr pReadH; TdFilePtr pReadH;
}; };
int32_t tsdbDelFReaderOpen(SDelFReader *pReader, SDelFile *pFile) { int32_t tsdbDelFReaderOpen(SDelFReader **ppReader, SDelFile *pFile) {
int32_t code = 0; int32_t code = 0;
// TODO // TODO
return code; return code;

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

@ -51,4 +51,26 @@ int32_t tsdbKeyCmprFn(const void *p1, const void *p2) {
} }
return 0; return 0;
}
int32_t tPutSDelIdx(uint8_t *p, SDelIdx *pDelIdx) {
int32_t n = 0;
n += tPutU32(p ? p + n : p, pDelIdx->delimiter);
n += tPutU8(p ? p + n : p, pDelIdx->flags);
n += tPutBinary(p ? p + n : p, pDelIdx->pOffset, pDelIdx->nOffset);
n += tPutBinary(p ? p + n : p, pDelIdx->pData, pDelIdx->nData);
return n;
}
int32_t tGetSDelIdx(uint8_t *p, SDelIdx *pDelIdx) {
int32_t n = 0;
n += tGetU32(p + n, &pDelIdx->delimiter);
n += tGetU8(p + n, &pDelIdx->flags);
n += tGetBinary(p + n, &pDelIdx->pOffset, &pDelIdx->nOffset);
n += tGetBinary(p + n, &pDelIdx->pData, &pDelIdx->nData);
return n;
} }