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

more coe

This commit is contained in:
Hongze Cheng 2022-12-01 11:45:26 +08:00
parent 16cd9db428
commit fc50d94156
2 changed files with 68 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -70,6 +70,9 @@ typedef struct SDiskData SDiskData;
typedef struct SDiskDataBuilder SDiskDataBuilder; typedef struct SDiskDataBuilder SDiskDataBuilder;
typedef struct SBlkInfo SBlkInfo; typedef struct SBlkInfo SBlkInfo;
#define TSDBROW_ROW_FMT ((int8_t)0x0)
#define TSDBROW_COL_FMT ((int8_t)0x1)
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F) #define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_MAX_SUBBLOCKS 8 #define TSDB_MAX_SUBBLOCKS 8
#define TSDB_FHDR_SIZE 512 #define TSDB_FHDR_SIZE 512
@ -102,12 +105,13 @@ static FORCE_INLINE int64_t tsdbLogicToFileSize(int64_t lSize, int32_t szPage) {
// tsdbUtil.c ============================================================================================== // tsdbUtil.c ==============================================================================================
// TSDBROW // TSDBROW
#define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow]) #define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow])
#define TSDBROW_VERSION(ROW) (((ROW)->type == 0) ? (ROW)->version : (ROW)->pBlockData->aVersion[(ROW)->iRow]) #define TSDBROW_VERSION(ROW) (((ROW)->type == 0) ? (ROW)->version : (ROW)->pBlockData->aVersion[(ROW)->iRow])
#define TSDBROW_SVERSION(ROW) TD_ROW_SVER((ROW)->pTSRow) #define TSDBROW_SVERSION(ROW) TD_ROW_SVER((ROW)->pTSRow)
#define TSDBROW_KEY(ROW) ((TSDBKEY){.version = TSDBROW_VERSION(ROW), .ts = TSDBROW_TS(ROW)}) #define TSDBROW_KEY(ROW) ((TSDBKEY){.version = TSDBROW_VERSION(ROW), .ts = TSDBROW_TS(ROW)})
#define tsdbRowFromTSRow(VERSION, TSROW) ((TSDBROW){.type = 0, .version = (VERSION), .pTSRow = (TSROW)}) #define tsdbRowFromTSRow(VERSION, TSROW) ((TSDBROW){.type = TSDBROW_ROW_FMT, .version = (VERSION), .pTSRow = (TSROW)})
#define tsdbRowFromBlockData(BLOCKDATA, IROW) ((TSDBROW){.type = 1, .pBlockData = (BLOCKDATA), .iRow = (IROW)}) #define tsdbRowFromBlockData(BLOCKDATA, IROW) \
((TSDBROW){.type = TSDBROW_COL_FMT, .pBlockData = (BLOCKDATA), .iRow = (IROW)})
void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal); void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
// int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow); // int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
int32_t tsdbRowCmprFn(const void *p1, const void *p2); int32_t tsdbRowCmprFn(const void *p1, const void *p2);
@ -335,10 +339,13 @@ struct SVersionRange {
typedef struct SMemSkipListNode SMemSkipListNode; typedef struct SMemSkipListNode SMemSkipListNode;
struct SMemSkipListNode { struct SMemSkipListNode {
int8_t level; int8_t level;
int8_t flag; // TSDBROW_ROW_FMT for row format, TSDBROW_COL_FMT for col format
int32_t iRow;
int64_t version; int64_t version;
SRow *pTSRow; void *pData;
SMemSkipListNode *forwards[0]; SMemSkipListNode *forwards[0];
}; };
typedef struct SMemSkipList { typedef struct SMemSkipList {
int64_t size; int64_t size;
uint32_t seed; uint32_t seed;
@ -376,7 +383,7 @@ struct SMemTable {
}; };
struct TSDBROW { struct TSDBROW {
int8_t type; // 0 for row from tsRow, 1 for row from block data int8_t type; // TSDBROW_ROW_FMT for row from tsRow, TSDBROW_COL_FMT for row from block data
union { union {
struct { struct {
int64_t version; int64_t version;
@ -795,8 +802,13 @@ static FORCE_INLINE TSDBROW *tsdbTbDataIterGet(STbDataIter *pIter) {
} }
pIter->pRow = &pIter->row; pIter->pRow = &pIter->row;
pIter->pRow->version = pIter->pNode->version; if (pIter->pNode->flag == TSDBROW_ROW_FMT) {
pIter->pRow->pTSRow = pIter->pNode->pTSRow; pIter->row = tsdbRowFromTSRow(pIter->pNode->version, pIter->pNode->pData);
} else if (pIter->pNode->flag == TSDBROW_COL_FMT) {
pIter->row = tsdbRowFromBlockData(pIter->pNode->pData, pIter->pNode->iRow);
} else {
ASSERT(0);
}
return pIter->pRow; return pIter->pRow;
} }

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

@ -247,7 +247,6 @@ void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDa
pIter->pTbData = pTbData; pIter->pTbData = pTbData;
pIter->backward = backward; pIter->backward = backward;
pIter->pRow = NULL; pIter->pRow = NULL;
pIter->row.type = 0;
if (pFrom == NULL) { if (pFrom == NULL) {
// create from head or tail // create from head or tail
if (backward) { if (backward) {
@ -415,8 +414,13 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) { for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_BACKWARD(px, iLevel); pn = SL_NODE_BACKWARD(px, iLevel);
while (pn != pTbData->sl.pHead) { while (pn != pTbData->sl.pHead) {
tKey.version = pn->version; if (pn->flag == TSDBROW_ROW_FMT) {
tKey.ts = pn->pTSRow->ts; tKey.version = pn->version;
tKey.ts = ((SRow *)pn->pData)->ts;
} else if (pn->flag == TSDBROW_COL_FMT) {
tKey.version = ((SBlockData *)pn->pData)->aVersion[pn->iRow];
tKey.ts = ((SBlockData *)pn->pData)->aTSKEY[pn->iRow];
}
int32_t c = tsdbKeyCmprFn(&tKey, pKey); int32_t c = tsdbKeyCmprFn(&tKey, pKey);
if (c <= 0) { if (c <= 0) {
@ -445,8 +449,13 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) { for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_FORWARD(px, iLevel); pn = SL_NODE_FORWARD(px, iLevel);
while (pn != pTbData->sl.pTail) { while (pn != pTbData->sl.pTail) {
tKey.version = pn->version; if (pn->flag == TSDBROW_ROW_FMT) {
tKey.ts = pn->pTSRow->ts; tKey.version = pn->version;
tKey.ts = ((SRow *)pn->pData)->ts;
} else if (pn->flag == TSDBROW_COL_FMT) {
tKey.version = ((SBlockData *)pn->pData)->aVersion[pn->iRow];
tKey.ts = ((SBlockData *)pn->pData)->aTSKEY[pn->iRow];
}
int32_t c = tsdbKeyCmprFn(&tKey, pKey); int32_t c = tsdbKeyCmprFn(&tKey, pKey);
if (c >= 0) { if (c >= 0) {
@ -473,29 +482,37 @@ static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
return level; return level;
} }
static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, int64_t version, SRow *pRow, static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, TSDBROW *pRow,
int8_t forward) { int8_t forward) {
int32_t code = 0; int32_t code = 0;
int8_t level; int8_t level;
SMemSkipListNode *pNode; SMemSkipListNode *pNode;
SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse; SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
ASSERT(pPool != NULL);
// node // node
level = tsdbMemSkipListRandLevel(&pTbData->sl); level = tsdbMemSkipListRandLevel(&pTbData->sl);
ASSERT(pPool != NULL);
pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level)); pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level));
if (pNode == NULL) { if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit; goto _exit;
} }
pNode->level = level; pNode->level = level;
pNode->version = version; pNode->flag = pRow->type;
pNode->pTSRow = vnodeBufPoolMalloc(pPool, pRow->len);
if (NULL == pNode->pTSRow) { if (pRow->type == TSDBROW_ROW_FMT) {
code = TSDB_CODE_OUT_OF_MEMORY; pNode->version = pRow->version;
goto _exit; pNode->pData = vnodeBufPoolMalloc(pPool, pRow->pTSRow->len);
if (NULL == pNode->pData) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
memcpy(pNode->pData, pRow->pTSRow, pRow->pTSRow->len);
} else if (pRow->type == TSDBROW_COL_FMT) {
pNode->iRow = pRow->iRow;
pNode->pData = pRow->pBlockData;
} }
memcpy(pNode->pTSRow, pRow, pRow->len);
for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) { for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) {
SMemSkipListNode *pn = pos[iLevel]; SMemSkipListNode *pn = pos[iLevel];
@ -589,13 +606,12 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData,
TSDBKEY key = {.version = version}; TSDBKEY key = {.version = version};
SMemSkipListNode *pos[SL_MAX_LEVEL]; SMemSkipListNode *pos[SL_MAX_LEVEL];
int32_t iRow = 0; int32_t iRow = 0;
TSDBROW row = tsdbRowFromBlockData(pBlockData, iRow); TSDBROW tRow = {.type = TSDBROW_COL_FMT, .pBlockData = pBlockData};
// TODO tRow.iRow = iRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD);
code = tbDataDoPut(pMemTable, pTbData, pos, version, NULL, 0); code = tbDataDoPut(pMemTable, pTbData, pos, &tRow, 0);
if (code) goto _exit; if (code) goto _exit;
++iRow;
if (iRow < pBlockData->nRow) { if (iRow < pBlockData->nRow) {
for (int8_t iLevel = pos[0]->level; iLevel < pTbData->sl.maxLevel; iLevel++) { for (int8_t iLevel = pos[0]->level; iLevel < pTbData->sl.maxLevel; iLevel++) {
@ -607,9 +623,9 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData,
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
} }
code = tbDataDoPut(pMemTable, pTbData, pos, version, NULL, 0); tRow.iRow = ++iRow;
code = tbDataDoPut(pMemTable, pTbData, pos, &tRow, 0);
if (code) goto _exit; if (code) goto _exit;
++iRow;
} }
} }
@ -621,25 +637,25 @@ static int32_t tsdbInsertRowDataToTable(SMemTable *pMemTable, STbData *pTbData,
SSubmitTbData *pSubmitTbData, int32_t *affectedRows) { SSubmitTbData *pSubmitTbData, int32_t *affectedRows) {
int32_t code = 0; int32_t code = 0;
SRow **rows = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP); int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP);
SRow **aRow = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP);
TSDBKEY key = {.version = version}; TSDBKEY key = {.version = version};
SMemSkipListNode *pos[SL_MAX_LEVEL]; SMemSkipListNode *pos[SL_MAX_LEVEL];
TSDBROW row = tsdbRowFromTSRow(version, NULL); TSDBROW tRow = {.type = TSDBROW_ROW_FMT, .version = version};
int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP);
int32_t iRow = 0; int32_t iRow = 0;
SRow *pLastRow = NULL; SRow *pLastRow = NULL;
// backward put first data // backward put first data
row.pTSRow = rows[iRow]; tRow.pTSRow = aRow[iRow];
key.ts = row.pTSRow->ts; key.ts = aRow[iRow]->ts;
iRow++; iRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD);
code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 0); code = tbDataDoPut(pMemTable, pTbData, pos, &tRow, 0);
if (code) goto _exit; if (code) goto _exit;
pTbData->minKey = TMIN(pTbData->minKey, key.ts); pTbData->minKey = TMIN(pTbData->minKey, key.ts);
pLastRow = row.pTSRow; pLastRow = tRow.pTSRow;
// forward put rest data // forward put rest data
if (iRow < nRow) { if (iRow < nRow) {
@ -648,17 +664,17 @@ static int32_t tsdbInsertRowDataToTable(SMemTable *pMemTable, STbData *pTbData,
} }
while (iRow < nRow) { while (iRow < nRow) {
row.pTSRow = rows[iRow]; tRow.pTSRow = aRow[iRow];
key.ts = row.pTSRow->ts; key.ts = tRow.pTSRow->ts;
if (SL_NODE_FORWARD(pos[0], 0) != pTbData->sl.pTail) { if (SL_NODE_FORWARD(pos[0], 0) != pTbData->sl.pTail) {
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS); tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
} }
code = tbDataDoPut(pMemTable, pTbData, pos, version, row.pTSRow, 1); code = tbDataDoPut(pMemTable, pTbData, pos, &tRow, 1);
if (code) goto _exit; if (code) goto _exit;
pLastRow = row.pTSRow; pLastRow = tRow.pTSRow;
iRow++; iRow++;
} }