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

more last refact

This commit is contained in:
Hongze Cheng 2022-08-04 05:42:15 +00:00
parent f84f88371b
commit d1c7e3a531
4 changed files with 90 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -135,13 +135,14 @@ int32_t tGetColData(uint8_t *p, SColData *pColData);
#define tBlockDataFirstKey(PBLOCKDATA) TSDBROW_KEY(&tBlockDataFirstRow(PBLOCKDATA))
#define tBlockDataLastKey(PBLOCKDATA) TSDBROW_KEY(&tBlockDataLastRow(PBLOCKDATA))
int32_t tBlockDataInit(SBlockData *pBlockData);
void tBlockDataClear(SBlockData *pBlockData, int8_t deepClear);
void tBlockDataReset(SBlockData *pBlockData);
int32_t tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema);
int32_t tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema, int64_t suid, int64_t uid);
int32_t tBlockDataAddColData(SBlockData *pBlockData, int32_t iColData, SColData **ppColData);
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
int32_t tBlockDataCorrectSchema(SBlockData *pBlockData, SBlockData *pBlockDataFrom);
void tBlockDataClearData(SBlockData *pBlockData);
void tBlockDataClear(SBlockData *pBlockData, int8_t deepClear);
int32_t tBlockDataAddColData(SBlockData *pBlockData, int32_t iColData, SColData **ppColData);
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema);
int32_t tBlockDataMerge(SBlockData *pBlockData1, SBlockData *pBlockData2, SBlockData *pBlockData);
int32_t tBlockDataCopy(SBlockData *pBlockDataSrc, SBlockData *pBlockDataDest);
SColData *tBlockDataGetColDataByIdx(SBlockData *pBlockData, int32_t idx);
@ -468,13 +469,17 @@ struct SColData {
uint8_t *pData;
};
// (SBlockData){.suid = 0, .uid = 0}: block data not initialized
// (SBlockData){.suid = suid, .uid = uid}: block data for ONE child table int .data file
// (SBlockData){.suid = suid, .uid = 0}: block data for N child tables int .last file
// (SBlockData){.suid = 0, .uid = uid}: block data for 1 normal table int .last/.data file
struct SBlockData {
int64_t suid; // 0 means normal table data block
int64_t uid; // 0 means block data in .last file, others in .data file
int32_t nRow;
int64_t *aUid;
int64_t *aVersion;
TSKEY *aTSKEY;
int64_t suid; // 0 means normal table block data, otherwise child table block data
int64_t uid; // 0 means block data in .last file, otherwise in .data file
int32_t nRow; // number of rows
int64_t *aUid; // uids of each row, only exist in block data in .last file (uid == 0)
int64_t *aVersion; // versions of each row
TSKEY *aTSKEY; // timestamp of each row
SArray *aIdx; // SArray<int32_t>
SArray *aColData; // SArray<SColData>
};

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

@ -296,22 +296,28 @@ static int32_t tsdbCommitterNextLastRow(SCommitter *pCommitter) {
ASSERT(pCommitter->dReader.pReader);
ASSERT(pCommitter->dReader.pRowInfo);
SBlockData *pBlockDatal = &pCommitter->dReader.bDatal;
pCommitter->dReader.iRow++;
if (pCommitter->dReader.iRow < pCommitter->dReader.bDatal.nRow) {
pCommitter->dReader.pRowInfo->uid = pCommitter->dReader.bData.aUid[pCommitter->dReader.iRow];
pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(&pCommitter->dReader.bDatal, pCommitter->dReader.iRow);
if (pCommitter->dReader.iRow < pBlockDatal->nRow) {
if (pBlockDatal->uid == 0) {
pCommitter->dReader.pRowInfo->uid = pBlockDatal->aUid[pCommitter->dReader.iRow];
}
pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(pBlockDatal, pCommitter->dReader.iRow);
} else {
pCommitter->dReader.iBlockL++;
if (pCommitter->dReader.iBlockL < taosArrayGetSize(pCommitter->dReader.aBlockL)) {
pCommitter->dReader.pBlockL = (SBlockL *)taosArrayGet(pCommitter->dReader.aBlockL, pCommitter->dReader.iBlockL);
code = tsdbReadLastBlock(pCommitter->dReader.pReader, pCommitter->dReader.pBlockL, &pCommitter->dReader.bDatal,
NULL, NULL);
code = tsdbReadLastBlock(pCommitter->dReader.pReader, pCommitter->dReader.pBlockL, pBlockDatal, NULL, NULL);
if (code) goto _exit;
pCommitter->dReader.iRow = 0;
pCommitter->dReader.pRowInfo->suid = pCommitter->dReader.pBlockL->suid;
pCommitter->dReader.pRowInfo->uid = pCommitter->dReader.bData.aUid[pCommitter->dReader.iRow];
pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(&pCommitter->dReader.bDatal, pCommitter->dReader.iRow);
pCommitter->dReader.pRowInfo->suid = pBlockDatal->suid;
if (pBlockDatal->uid) {
pCommitter->dReader.pRowInfo->uid = pBlockDatal->uid;
} else {
pCommitter->dReader.pRowInfo->uid = pBlockDatal->aUid[0];
}
pCommitter->dReader.pRowInfo->row = tsdbRowFromBlockData(pBlockDatal, pCommitter->dReader.iRow);
} else {
pCommitter->dReader.pRowInfo = NULL;
}
@ -354,15 +360,16 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
} else {
pCommitter->dReader.pBlockIdx = NULL;
}
tBlockDataReset(&pCommitter->dReader.bData);
// last
code = tsdbReadBlockL(pCommitter->dReader.pReader, pCommitter->dReader.aBlockL, NULL);
if (code) goto _err;
pCommitter->dReader.iBlockL = -1;
pCommitter->dReader.bDatal.nRow = 0;
pCommitter->dReader.iRow = -1;
pCommitter->dReader.pRowInfo = &pCommitter->dReader.rowInfo;
tBlockDataReset(&pCommitter->dReader.bDatal);
code = tsdbCommitterNextLastRow(pCommitter);
if (code) goto _err;
} else {

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

@ -591,8 +591,8 @@ int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx, uint8_t **ppB
ASSERT(n + sizeof(TSCKSUM) == size);
_exit:
tFree(pBuf);
_exit:
return code;
_err:
@ -658,8 +658,8 @@ int32_t tsdbReadBlockL(SDataFReader *pReader, SArray *aBlockL, uint8_t **ppBuf)
ASSERT(n + sizeof(TSCKSUM) == size);
_exit:
tFree(pBuf);
_exit:
return code;
_err:

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

@ -1121,6 +1121,8 @@ static FORCE_INLINE int32_t tColDataCmprFn(const void *p1, const void *p2) {
int32_t tBlockDataInit(SBlockData *pBlockData) {
int32_t code = 0;
pBlockData->suid = 0;
pBlockData->uid = 0;
pBlockData->nRow = 0;
pBlockData->aUid = NULL;
pBlockData->aVersion = NULL;
@ -1141,37 +1143,46 @@ _exit:
return code;
}
void tBlockDataReset(SBlockData *pBlockData) {
pBlockData->nRow = 0;
taosArrayClear(pBlockData->aIdx);
}
void tBlockDataClear(SBlockData *pBlockData, int8_t deepClear) {
tFree((uint8_t *)pBlockData->aUid);
tFree((uint8_t *)pBlockData->aVersion);
tFree((uint8_t *)pBlockData->aTSKEY);
taosArrayDestroy(pBlockData->aIdx);
taosArrayDestroyEx(pBlockData->aColData, deepClear ? tColDataClear : NULL);
pBlockData->aColData = NULL;
pBlockData->aIdx = NULL;
pBlockData->aTSKEY = NULL;
pBlockData->aUid = NULL;
pBlockData->aVersion = NULL;
pBlockData->aTSKEY = NULL;
pBlockData->aIdx = NULL;
pBlockData->aColData = NULL;
}
int32_t tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema) {
void tBlockDataReset(SBlockData *pBlockData) {
pBlockData->suid = 0;
pBlockData->uid = 0;
pBlockData->nRow = 0;
taosArrayClear(pBlockData->aIdx);
}
int32_t tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema, int64_t suid, int64_t uid) {
int32_t code = 0;
SColData *pColData;
STColumn *pTColumn;
ASSERT(suid || uid);
tBlockDataReset(pBlockData);
pBlockData->suid = suid;
pBlockData->uid = uid;
if (pTSchema) {
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
pTColumn = &pTSchema->columns[iColumn];
STColumn *pTColumn = &pTSchema->columns[iColumn];
SColData *pColData;
code = tBlockDataAddColData(pBlockData, iColumn - 1, &pColData);
if (code) goto _exit;
tColDataInit(pColData, pTColumn->colId, pTColumn->type, (pTColumn->flags & COL_SMA_ON) != 0);
}
}
_exit:
return code;
@ -1211,52 +1222,47 @@ _err:
return code;
}
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema) {
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid) {
int32_t code = 0;
// TSDBKEY
ASSERT(pBlockData->suid || pBlockData->uid);
// uid
if (pBlockData->uid == 0) {
ASSERT(uid);
code = tRealloc((uint8_t **)&pBlockData->aUid, sizeof(int64_t) * (pBlockData->nRow + 1));
if (code) goto _err;
pBlockData->aUid[pBlockData->nRow] = uid;
}
// version
code = tRealloc((uint8_t **)&pBlockData->aVersion, sizeof(int64_t) * (pBlockData->nRow + 1));
if (code) goto _err;
pBlockData->aVersion[pBlockData->nRow] = TSDBROW_VERSION(pRow);
// timestamp
code = tRealloc((uint8_t **)&pBlockData->aTSKEY, sizeof(TSKEY) * (pBlockData->nRow + 1));
if (code) goto _err;
pBlockData->aVersion[pBlockData->nRow] = TSDBROW_VERSION(pRow);
pBlockData->aTSKEY[pBlockData->nRow] = TSDBROW_TS(pRow);
// OTHER
int32_t iColData = 0;
int32_t nColData = taosArrayGetSize(pBlockData->aIdx);
SRowIter iter = {0};
SRowIter *pIter = &iter;
SColData *pColData;
SRowIter rIter = {0};
SColVal *pColVal;
if (nColData == 0) goto _exit;
tRowIterInit(&rIter, pRow, pTSchema);
pColVal = tRowIterNext(&rIter);
for (int32_t iColData = 0; iColData < taosArrayGetSize(pBlockData->aIdx); iColData++) {
SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
tRowIterInit(pIter, pRow, pTSchema);
pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
pColVal = tRowIterNext(pIter);
while (pColVal && pColVal->cid < pColData->cid) {
pColVal = tRowIterNext(&rIter);
}
while (pColData) {
if (pColVal) {
if (pColData->cid == pColVal->cid) {
if (pColVal == NULL || pColVal->cid > pColData->cid) {
code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
if (code) goto _err;
} else {
code = tColDataAppendValue(pColData, pColVal);
if (code) goto _err;
pColVal = tRowIterNext(pIter);
pColData = ((++iColData) < nColData) ? tBlockDataGetColDataByIdx(pBlockData, iColData) : NULL;
} else if (pColData->cid < pColVal->cid) {
code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
if (code) goto _err;
pColData = ((++iColData) < nColData) ? tBlockDataGetColDataByIdx(pBlockData, iColData) : NULL;
} else {
pColVal = tRowIterNext(pIter);
}
} else {
code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
if (code) goto _err;
pColData = ((++iColData) < nColData) ? tBlockDataGetColDataByIdx(pBlockData, iColData) : NULL;
pColVal = tRowIterNext(&rIter);
}
}