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

feat: tsdb multi-version 1

This commit is contained in:
Hongze Cheng 2022-06-06 10:00:48 +00:00
parent 3b8e93f48a
commit 487530636d
8 changed files with 702 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -39,35 +39,46 @@ typedef struct SDelOp SDelOp;
static int tsdbKeyCmprFn(const void *p1, const void *p2); static int tsdbKeyCmprFn(const void *p1, const void *p2);
// tsdbMemTable ================ // tsdbMemTable ==============================================================================================
typedef struct STbData STbData; typedef struct STbData STbData;
typedef struct SMemTable SMemTable; typedef struct SMemTable SMemTable;
typedef struct SMergeInfo SMergeInfo; typedef struct STbDataIter STbDataIter;
typedef struct STable STable; typedef struct SMergeInfo SMergeInfo;
typedef struct STable STable;
int tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable); // SMemTable
void tsdbMemTableDestroy(SMemTable *pMemTable); int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, void tsdbMemTableDestroy(SMemTable *pMemTable);
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo); void tsdbGetTbDataFromMemTable(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData);
// STbDataIter
int32_t tsdbTbDataIterCreate(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter **ppIter);
void *tsdbTbDataIterDestroy(STbDataIter *pIter);
void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter *pIter);
bool tsdbTbDataIterNext(STbDataIter *pIter);
bool tsdbTbDataIterGet(STbDataIter *pIter, TSDBROW *pRow);
int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
// tsdbMemTable2.c ============================================================================================== // tsdbMemTable2.c ==============================================================================================
typedef struct SMemTable2 SMemTable2; // typedef struct SMemTable2 SMemTable2;
typedef struct SMemData SMemData; // typedef struct SMemData SMemData;
typedef struct SMemDataIter SMemDataIter; // typedef struct SMemDataIter SMemDataIter;
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable2 **ppMemTable); // int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable2 **ppMemTable);
void tsdbMemTableDestroy2(SMemTable2 *pMemTable); // void tsdbMemTableDestroy2(SMemTable2 *pMemTable);
int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmitBlk); // int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmitBlk);
int32_t tsdbDeleteTableData2(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey); // int32_t tsdbDeleteTableData2(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
/* SMemDataIter */ // /* SMemDataIter */
void tsdbMemDataIterOpen(SMemData *pMemData, TSDBKEY *pKey, int8_t backward, SMemDataIter *pIter); // void tsdbMemDataIterOpen(SMemData *pMemData, TSDBKEY *pKey, int8_t backward, SMemDataIter *pIter);
bool tsdbMemDataIterNext(SMemDataIter *pIter); // bool tsdbMemDataIterNext(SMemDataIter *pIter);
void tsdbMemDataIterGet(SMemDataIter *pIter, TSDBROW **ppRow); // void tsdbMemDataIterGet(SMemDataIter *pIter, TSDBROW **ppRow);
// tsdbCommit2.c ============================================================================================== // // tsdbCommit2.c ==============================================================================================
int32_t tsdbBegin2(STsdb *pTsdb); // int32_t tsdbBegin2(STsdb *pTsdb);
int32_t tsdbCommit2(STsdb *pTsdb); // int32_t tsdbCommit2(STsdb *pTsdb);
// tsdbFile.c ============================================================================================== // tsdbFile.c ==============================================================================================
typedef int32_t TSDB_FILE_T; typedef int32_t TSDB_FILE_T;
@ -211,29 +222,39 @@ struct TSDBKEY {
TSKEY ts; TSKEY ts;
}; };
typedef struct SMemSkipListNode SMemSkipListNode;
struct SMemSkipListNode {
int8_t level;
SMemSkipListNode *forwards[0];
};
typedef struct SMemSkipList {
uint32_t seed;
int64_t size;
int8_t maxLevel;
int8_t level;
SMemSkipListNode *pHead;
SMemSkipListNode *pTail;
} SMemSkipList;
struct STbData { struct STbData {
tb_uid_t suid; tb_uid_t suid;
tb_uid_t uid; tb_uid_t uid;
TSDBKEY minKey; TSDBKEY minKey;
TSDBKEY maxKey; TSDBKEY maxKey;
SDelOp *pHead; SDelOp *pHead;
SDelOp *pTail; SDelOp *pTail;
int64_t nrows; SMemSkipList sl;
SSkipList *pData;
}; };
struct SMemTable { struct SMemTable {
STsdb *pTsdb; SRWLatch latch;
int32_t nRef; STsdb *pTsdb;
SRWLatch latch; int32_t nRef;
TSDBKEY minKey; TSDBKEY minKey;
TSDBKEY maxKey; TSDBKEY maxKey;
int64_t nRow; int64_t nRow;
int64_t nDelOp; int64_t nDelOp;
SDelOp *pHead; SArray *aTbData; // SArray<STbData>
SDelOp *pTail;
SSkipList *pSlIdx; // SSkiplist<STbData>
SHashObj *pHashIdx;
}; };
struct STsdbFSMeta { struct STsdbFSMeta {
@ -656,7 +677,7 @@ struct SFSIter {
struct TSDBROW { struct TSDBROW {
int64_t version; int64_t version;
STSRow2 tsRow; STSRow *pTSRow;
}; };
struct TABLEID { struct TABLEID {
@ -709,16 +730,6 @@ static FORCE_INLINE int tsdbKeyCmprFn(const void *p1, const void *p2) {
return 0; return 0;
} }
typedef struct SMemSkipListNode SMemSkipListNode;
typedef struct SMemSkipList {
uint32_t seed;
int32_t size;
int8_t maxLevel;
int8_t level;
SMemSkipListNode *pHead;
SMemSkipListNode *pTail;
} SMemSkipList;
struct SMemData { struct SMemData {
tb_uid_t suid; tb_uid_t suid;
tb_uid_t uid; tb_uid_t uid;
@ -730,13 +741,19 @@ struct SMemData {
}; };
struct SMemDataIter { struct SMemDataIter {
SMemData *pMemData; STbData *pMemData;
int8_t backward; int8_t backward;
TSDBROW *pRow; TSDBROW *pRow;
SMemSkipListNode *pNode; // current node SMemSkipListNode *pNode; // current node
TSDBROW row; TSDBROW row;
}; };
struct STbDataIter {
STbData *pTbData;
int8_t backward;
SMemSkipListNode *pNode;
};
#endif #endif
#ifdef __cplusplus #ifdef __cplusplus

4
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -116,7 +116,9 @@ int tsdbBegin(STsdb* pTsdb);
int32_t tsdbCommit(STsdb* pTsdb); int32_t tsdbCommit(STsdb* pTsdb);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg); int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp); int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
int tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp); int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock,
SSubmitBlkRsp* pRsp);
int32_t tsdbDeleteTableData(STsdb* pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId, tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId,
uint64_t taskId); uint64_t taskId);
tsdbReaderT tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId, tsdbReaderT tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId,

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

@ -58,26 +58,26 @@ typedef struct {
#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->pVnode->config.tsdbCfg.maxRows) #define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->pVnode->config.tsdbCfg.maxRows)
#define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch))) #define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
static void tsdbStartCommit(STsdb *pRepo); static void tsdbStartCommit(STsdb *pRepo);
static void tsdbEndCommit(STsdb *pTsdb, int eno); static void tsdbEndCommit(STsdb *pTsdb, int eno);
static int tsdbInitCommitH(SCommitH *pCommith, STsdb *pRepo); static int tsdbInitCommitH(SCommitH *pCommith, STsdb *pRepo);
static void tsdbSeekCommitIter(SCommitH *pCommith, TSKEY key); static void tsdbSeekCommitIter(SCommitH *pCommith, TSKEY key);
static int tsdbNextCommitFid(SCommitH *pCommith); static int tsdbNextCommitFid(SCommitH *pCommith);
static void tsdbDestroyCommitH(SCommitH *pCommith); static void tsdbDestroyCommitH(SCommitH *pCommith);
static int tsdbCreateCommitIters(SCommitH *pCommith); static int32_t tsdbCreateCommitIters(SCommitH *pCommith);
static void tsdbDestroyCommitIters(SCommitH *pCommith); static void tsdbDestroyCommitIters(SCommitH *pCommith);
static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid); static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
static void tsdbResetCommitFile(SCommitH *pCommith); static void tsdbResetCommitFile(SCommitH *pCommith);
static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid); static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
static int tsdbCommitToTable(SCommitH *pCommith, int tid); static int tsdbCommitToTable(SCommitH *pCommith, int tid);
static bool tsdbCommitIsSameFile(SCommitH *pCommith, int bidx); static bool tsdbCommitIsSameFile(SCommitH *pCommith, int bidx);
static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx); static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx);
static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable); static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable);
static int tsdbComparKeyBlock(const void *arg1, const void *arg2); static int tsdbComparKeyBlock(const void *arg1, const void *arg2);
static int tsdbWriteBlockInfo(SCommitH *pCommih); static int tsdbWriteBlockInfo(SCommitH *pCommih);
static int tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLimit, bool toData); static int tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLimit, bool toData);
static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx); static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx);
static int tsdbMoveBlock(SCommitH *pCommith, int bidx); static int tsdbMoveBlock(SCommitH *pCommith, int bidx);
static int tsdbCommitAddBlock(SCommitH *pCommith, const SBlock *pSupBlock, const SBlock *pSubBlocks, int nSubBlocks); static int tsdbCommitAddBlock(SCommitH *pCommith, const SBlock *pSupBlock, const SBlock *pSubBlocks, int nSubBlocks);
static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols *pDataCols, TSKEY keyLimit, static int tsdbMergeBlockData(SCommitH *pCommith, SCommitIter *pIter, SDataCols *pDataCols, TSKEY keyLimit,
bool isLastOneBlock); bool isLastOneBlock);
@ -453,11 +453,32 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
return 0; return 0;
} }
static int tsdbCreateCommitIters(SCommitH *pCommith) { static int32_t tsdbCreateCommitIters(SCommitH *pCommith) {
STsdb *pRepo = TSDB_COMMIT_REPO(pCommith); int32_t code = 0;
SMemTable *pMem = pRepo->imem; STsdb *pRepo = TSDB_COMMIT_REPO(pCommith);
SMemTable *pMem = pRepo->imem;
SCommitIter *pCommitIter;
pCommith->niters = taosArrayGetSize(pMem->aTbData);
pCommith->iters = (SCommitIter *)taosMemoryCalloc(pCommith->niters, sizeof(SCommitIter));
if (pCommith->iters == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
for (int32_t iIter = 0; iIter < pCommith->niters; iIter++) {
pCommitIter = (SCommitIter *)taosArrayGetP(pMem->aTbData, iIter);
// TODO
// pCommitIter->pIter =
}
return code;
_err:
return code;
#if 0
SSkipListIterator *pSlIter; SSkipListIterator *pSlIter;
SCommitIter *pCommitIter;
SSkipListNode *pNode; SSkipListNode *pNode;
STbData *pTbData; STbData *pTbData;
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
@ -495,8 +516,7 @@ static int tsdbCreateCommitIters(SCommitH *pCommith) {
} }
} }
tSkipListDestroyIter(pSlIter); tSkipListDestroyIter(pSlIter);
#endif
return 0;
} }
static void tsdbDestroyCommitIters(SCommitH *pCommith) { static void tsdbDestroyCommitIters(SCommitH *pCommith) {

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

@ -15,69 +15,301 @@
#include "tsdb.h" #include "tsdb.h"
static STbData *tsdbNewTbData(tb_uid_t uid); #define SL_MAX_LEVEL 5
static void tsdbFreeTbData(STbData *pTbData);
static char *tsdbGetTsTupleKey(const void *data);
static int tsdbTbDataComp(const void *arg1, const void *arg2);
static char *tsdbTbDataGetUid(const void *arg);
static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row,
bool merge);
int tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) { #define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
SMemTable *pMemTable; #define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
SVnode *pVnode; #define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level))
*ppMemTable = NULL; #define SL_MOVE_BACKWARD 0x1
pVnode = pTsdb->pVnode; #define SL_MOVE_FROM_POS 0x2
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow);
static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags);
static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData);
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp);
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable) {
int32_t code = 0;
SMemTable *pMemTable = NULL;
// alloc handle
pMemTable = (SMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable)); pMemTable = (SMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable));
if (pMemTable == NULL) { if (pMemTable == NULL) {
return -1; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
} }
taosInitRWLatch(&pMemTable->latch);
pMemTable->pTsdb = pTsdb; pMemTable->pTsdb = pTsdb;
pMemTable->nRef = 1; pMemTable->nRef = 1;
taosInitRWLatch(&pMemTable->latch);
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->pSlIdx = tSkipListCreate(pVnode->config.tsdbCfg.slLevel, TSDB_DATA_TYPE_BIGINT, sizeof(tb_uid_t), pMemTable->nDelOp = 0;
tsdbTbDataComp, SL_DISCARD_DUP_KEY, tsdbTbDataGetUid); pMemTable->aTbData = taosArrayInit(128, sizeof(STbData *));
if (pMemTable->pSlIdx == NULL) { if (pMemTable->aTbData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pMemTable); taosMemoryFree(pMemTable);
return -1; goto _err;
}
pMemTable->pHashIdx = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (pMemTable->pHashIdx == NULL) {
tSkipListDestroy(pMemTable->pSlIdx);
taosMemoryFree(pMemTable);
return -1;
} }
*ppMemTable = pMemTable; *ppMemTable = pMemTable;
return 0; return code;
_err:
*ppMemTable = NULL;
return code;
} }
void tsdbMemTableDestroy(SMemTable *pMemTable) { void tsdbMemTableDestroy(SMemTable *pMemTable) {
if (pMemTable) { if (pMemTable) {
taosHashCleanup(pMemTable->pHashIdx); taosArrayDestroy(pMemTable->aTbData);
SSkipListIterator *pIter = tSkipListCreateIter(pMemTable->pSlIdx);
SSkipListNode *pNode = NULL;
STbData *pTbData = NULL;
for (;;) {
if (!tSkipListIterNext(pIter)) break;
pNode = tSkipListIterGet(pIter);
pTbData = (STbData *)pNode->pData;
tsdbFreeTbData(pTbData);
}
tSkipListDestroyIter(pIter);
tSkipListDestroy(pMemTable->pSlIdx);
taosMemoryFree(pMemTable); taosMemoryFree(pMemTable);
} }
} }
static int32_t tbDataPCmprFn(const void *p1, const void *p2) {
STbData *pTbData1 = *(STbData **)p1;
STbData *pTbData2 = *(STbData **)p2;
if (pTbData1->suid < pTbData2->suid) {
return -1;
} else if (pTbData1->suid > pTbData2->suid) {
return 1;
}
if (pTbData1->uid < pTbData2->uid) {
return -1;
} else if (pTbData1->uid > pTbData2->uid) {
return 1;
}
return 0;
}
void tsdbGetTbDataFromMemTable(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData) {
STbData *pTbData = &(STbData){.suid = suid, .uid = uid};
void *p = taosArraySearch(pMemTable->aTbData, &pTbData, tbDataPCmprFn, TD_EQ);
*ppTbData = p ? *(STbData **)p : NULL;
}
int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock,
SSubmitBlkRsp *pRsp) {
int32_t code = 0;
SMemTable *pMemTable = pTsdb->mem;
STbData *pTbData = NULL;
tb_uid_t suid = pMsgIter->suid;
tb_uid_t uid = pMsgIter->uid;
int32_t sverNew;
// check if table exists (todo: refact)
SMetaReader mr = {0};
SMetaEntry me = {0};
metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0);
if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) {
metaReaderClear(&mr);
code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
goto _err;
}
if (pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);
if (mr.me.type == TSDB_NORMAL_TABLE) {
sverNew = mr.me.ntbEntry.schemaRow.version;
} else {
metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid);
sverNew = mr.me.stbEntry.schemaRow.version;
}
metaReaderClear(&mr);
pRsp->sver = sverNew;
// create/get STbData to op
code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData);
if (code) {
goto _err;
}
// do insert impl
code = tsdbInsertTableDataImpl(pMemTable, pTbData, version, pMsgIter, pBlock, pRsp);
if (code) {
goto _err;
}
return code;
_err:
return code;
}
int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey) {
int32_t code = 0;
SMemTable *pMemTable = pTsdb->mem;
STbData *pTbData = NULL;
SVBufPool *pPool = pTsdb->pVnode->inUse;
// check if table exists (todo)
code = tsdbGetOrCreateTbData(pMemTable, suid, uid, &pTbData);
if (code) {
goto _err;
}
// do delete
SDelOp *pDelOp = (SDelOp *)vnodeBufPoolMalloc(pPool, sizeof(*pDelOp));
if (pDelOp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pDelOp->version = version;
pDelOp->sKey = sKey;
pDelOp->eKey = eKey;
pDelOp->pNext = NULL;
if (pTbData->pHead == NULL) {
ASSERT(pTbData->pTail == NULL);
pTbData->pHead = pTbData->pTail = pDelOp;
} else {
pTbData->pTail->pNext = pDelOp;
pTbData->pTail = pDelOp;
}
// update the state of pMemTable and other (todo)
pMemTable->nDelOp++;
tsdbError("vgId:%d delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
" since %s",
TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
return code;
_err:
tsdbError("vgId:%d failed to delete data from table suid:%" PRId64 " uid:%" PRId64 " skey:%" PRId64 " eKey:%" PRId64
" since %s",
TD_VID(pTsdb->pVnode), suid, uid, sKey, eKey, tstrerror(code));
return code;
}
static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row,
bool merge) {
if (pCols) {
if (*ppSchema == NULL || schemaVersion(*ppSchema) != TD_ROW_SVER(row)) {
*ppSchema = tsdbGetTableSchemaImpl(pTsdb, pTable, false, false, TD_ROW_SVER(row));
if (*ppSchema == NULL) {
ASSERT(false);
return -1;
}
}
tdAppendSTSRowToDataCol(row, *ppSchema, pCols, merge);
}
return 0;
}
int32_t tsdbTbDataIterCreate(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter **ppIter) {
int32_t code = 0;
(*ppIter) = (STbDataIter *)taosMemoryCalloc(1, sizeof(STbDataIter));
if ((*ppIter) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
tsdbTbDataIterOpen(pTbData, pFrom, backward, *ppIter);
_exit:
return code;
}
void *tsdbTbDataIterDestroy(STbDataIter *pIter) {
if (pIter) {
taosMemoryFree(pIter);
}
return NULL;
}
void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDataIter *pIter) {
SMemSkipListNode *pos[SL_MAX_LEVEL];
pIter->pTbData = pTbData;
pIter->backward = backward;
if (pFrom == NULL) {
// create from head or tail
if (backward) {
pIter->pNode = SL_NODE_BACKWARD(pTbData->sl.pTail, 0);
} else {
pIter->pNode = SL_NODE_FORWARD(pTbData->sl.pHead, 0);
}
} else {
// create from a key
if (backward) {
tbDataMovePosTo(pTbData, pos, pFrom, SL_MOVE_BACKWARD);
pIter->pNode = SL_NODE_BACKWARD(pos[0], 0);
} else {
tbDataMovePosTo(pTbData, pos, pFrom, 0);
pIter->pNode = SL_NODE_FORWARD(pos[0], 0);
}
}
}
bool tsdbTbDataIterNext(STbDataIter *pIter) {
SMemSkipListNode *pHead = pIter->pTbData->sl.pHead;
SMemSkipListNode *pTail = pIter->pTbData->sl.pTail;
if (pIter->backward) {
ASSERT(pIter->pNode != pTail);
if (pIter->pNode == pHead) {
return false;
}
pIter->pNode = SL_NODE_BACKWARD(pIter->pNode, 0);
if (pIter->pNode == pHead) {
return false;
}
} else {
ASSERT(pIter->pNode != pHead);
if (pIter->pNode == pTail) {
return false;
}
pIter->pNode = SL_NODE_FORWARD(pIter->pNode, 0);
if (pIter->pNode == pTail) {
return false;
}
}
return true;
}
bool tsdbTbDataIterGet(STbDataIter *pIter, TSDBROW *pRow) {
SMemSkipListNode *pHead = pIter->pTbData->sl.pHead;
SMemSkipListNode *pTail = pIter->pTbData->sl.pTail;
TSDBROW row = {0};
if (pRow == NULL) {
pRow = &row;
}
if (pIter->backward) {
ASSERT(pIter->pNode != pTail);
if (pIter->pNode == pHead) {
return false;
}
} else {
ASSERT(pIter->pNode != pHead);
if (pIter->pNode == pTail) {
return false;
}
}
tGetTSDBRow((uint8_t *)SL_NODE_DATA(pIter->pNode), pRow);
return true;
}
/** /**
* This is an important function to load data or try to load data from memory skiplist iterator. * This is an important function to load data or try to load data from memory skiplist iterator.
* *
@ -238,151 +470,276 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter
return 0; return 0;
} }
int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) { static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData) {
SSubmitBlkIter blkIter = {0}; int32_t code = 0;
SMemTable *pMemTable = pTsdb->mem; int32_t idx = 0;
void *tptr; STbData *pTbData = NULL;
STbData *pTbData; STbData *pTbDataT = &(STbData){.suid = suid, .uid = uid};
STSRow *row;
TSKEY keyMin;
TSKEY keyMax;
SSubmitBlk *pBlkCopy;
int64_t sverNew;
// check if table exists // get
SMetaReader mr = {0}; idx = taosArraySearchIdx(pMemTable->aTbData, &pTbDataT, tbDataPCmprFn, TD_GE);
SMetaEntry me = {0}; if (idx >= 0) {
metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0); pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, idx);
if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) { if (tbDataPCmprFn(&pTbDataT, &pTbData) == 0) goto _exit;
metaReaderClear(&mr);
terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
return -1;
} }
if (pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);
if (mr.me.type == TSDB_NORMAL_TABLE) { // create
sverNew = mr.me.ntbEntry.schemaRow.version; SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
} else { int8_t maxLevel = pMemTable->pTsdb->pVnode->config.tsdbCfg.slLevel;
metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid);
sverNew = mr.me.stbEntry.schemaRow.version; pTbData = vnodeBufPoolMalloc(pPool, sizeof(*pTbData) + SL_NODE_SIZE(maxLevel) * 2);
if (pTbData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
} }
metaReaderClear(&mr); pTbData->suid = suid;
pTbData->uid = uid;
pTbData->minKey = (TSDBKEY){.ts = TSKEY_MAX, .version = INT64_MAX};
pTbData->maxKey = (TSDBKEY){.ts = TSKEY_MIN, .version = -1};
pTbData->pHead = NULL;
pTbData->pTail = NULL;
pTbData->sl.seed = taosRand();
pTbData->sl.size = 0;
pTbData->sl.maxLevel = maxLevel;
pTbData->sl.level = 0;
pTbData->sl.pHead = (SMemSkipListNode *)&pTbData[1];
pTbData->sl.pTail = (SMemSkipListNode *)POINTER_SHIFT(pTbData->sl.pHead, SL_NODE_SIZE(maxLevel));
pTbData->sl.pHead->level = maxLevel;
pTbData->sl.pTail->level = maxLevel;
for (int8_t iLevel = 0; iLevel < maxLevel; iLevel++) {
SL_NODE_FORWARD(pTbData->sl.pHead, iLevel) = pTbData->sl.pTail;
SL_NODE_BACKWARD(pTbData->sl.pTail, iLevel) = pTbData->sl.pHead;
// create container is nedd SL_NODE_BACKWARD(pTbData->sl.pHead, iLevel) = NULL;
tptr = taosHashGet(pMemTable->pHashIdx, &(pMsgIter->uid), sizeof(pMsgIter->uid)); SL_NODE_FORWARD(pTbData->sl.pTail, iLevel) = NULL;
if (tptr == NULL) { }
pTbData = tsdbNewTbData(pMsgIter->uid);
if (pTbData == NULL) { if (taosArrayInsert(pMemTable->aTbData, idx < 0 ? 0 : idx, &pTbData) == NULL) {
return -1; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
_exit:
*ppTbData = pTbData;
return code;
_err:
*ppTbData = NULL;
return code;
}
static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags) {
SMemSkipListNode *px;
SMemSkipListNode *pn;
TSDBKEY *pTKey;
int c;
int backward = flags & SL_MOVE_BACKWARD;
int fromPos = flags & SL_MOVE_FROM_POS;
if (backward) {
px = pTbData->sl.pTail;
for (int8_t iLevel = pTbData->sl.maxLevel - 1; iLevel >= pTbData->sl.level; iLevel--) {
pos[iLevel] = px;
} }
// Put into hash if (pTbData->sl.level) {
taosHashPut(pMemTable->pHashIdx, &(pMsgIter->uid), sizeof(pMsgIter->uid), &(pTbData), sizeof(pTbData)); if (fromPos) px = pos[pTbData->sl.level - 1];
// Put into skiplist for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
tSkipListPut(pMemTable->pSlIdx, pTbData); pn = SL_NODE_BACKWARD(px, iLevel);
} else { while (pn != pTbData->sl.pHead) {
pTbData = *(STbData **)tptr; pTKey = (TSDBKEY *)SL_NODE_DATA(pn);
}
// copy data to buffer pool c = tsdbKeyCmprFn(pTKey, pKey);
int32_t tlen = pMsgIter->dataLen + pMsgIter->schemaLen + sizeof(*pBlock); if (c <= 0) {
pBlkCopy = (SSubmitBlk *)vnodeBufPoolMalloc(pTsdb->pVnode->inUse, tlen); break;
memcpy(pBlkCopy, pBlock, tlen); } else {
px = pn;
pn = SL_NODE_BACKWARD(px, iLevel);
}
}
tInitSubmitBlkIter(pMsgIter, pBlkCopy, &blkIter); pos[iLevel] = px;
if (blkIter.row == NULL) return 0;
keyMin = TD_ROW_KEY(blkIter.row);
tSkipListPutBatchByIter(pTbData->pData, &blkIter, (iter_next_fn_t)tGetSubmitBlkNext);
#ifdef TD_DEBUG_PRINT_ROW
printf("!!! %s:%d vgId:%d dir:%s table:%" PRIi64 " has %d rows in skiplist\n\n", __func__, __LINE__,
TD_VID(pTsdb->pVnode), pTsdb->dir, pTbData->uid, SL_SIZE(pTbData->pData));
#endif
// Set statistics
keyMax = TD_ROW_KEY(blkIter.row);
pTbData->nrows += pMsgIter->numOfRows;
if (pTbData->minKey.ts > keyMin) pTbData->minKey.ts = keyMin;
if (pTbData->maxKey.ts < keyMax) pTbData->maxKey.ts = keyMax;
pMemTable->nRow += pMsgIter->numOfRows;
if (pMemTable->minKey.ts > keyMin) pMemTable->minKey.ts = keyMin;
if (pMemTable->maxKey.ts < keyMax) pMemTable->maxKey.ts = keyMax;
pRsp->numOfRows = pMsgIter->numOfRows;
pRsp->affectedRows = pMsgIter->numOfRows;
pRsp->sver = sverNew;
return 0;
}
static STbData *tsdbNewTbData(tb_uid_t uid) {
STbData *pTbData = (STbData *)taosMemoryCalloc(1, sizeof(*pTbData));
if (pTbData == NULL) {
return NULL;
}
pTbData->uid = uid;
pTbData->minKey.ts = TSKEY_MAX;
pTbData->maxKey.ts = TSKEY_MIN;
pTbData->nrows = 0;
#if 0
pTbData->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), tkeyComparFn, SL_DISCARD_DUP_KEY,
tsdbGetTsTupleKey);
#endif
pTbData->pData =
tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), tkeyComparFn, SL_ALLOW_DUP_KEY, tsdbGetTsTupleKey);
if (pTbData->pData == NULL) {
taosMemoryFree(pTbData);
return NULL;
}
return pTbData;
}
static void tsdbFreeTbData(STbData *pTbData) {
if (pTbData) {
tSkipListDestroy(pTbData->pData);
taosMemoryFree(pTbData);
}
}
static char *tsdbGetTsTupleKey(const void *data) { return (char *)TD_ROW_KEY_ADDR((STSRow *)data); }
static int tsdbTbDataComp(const void *arg1, const void *arg2) {
STbData *pTbData1 = (STbData *)arg1;
STbData *pTbData2 = (STbData *)arg2;
if (pTbData1->uid > pTbData2->uid) {
return 1;
} else if (pTbData1->uid == pTbData2->uid) {
return 0;
} else {
return -1;
}
}
static char *tsdbTbDataGetUid(const void *arg) {
STbData *pTbData = (STbData *)arg;
return (char *)(&(pTbData->uid));
}
static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row,
bool merge) {
if (pCols) {
if (*ppSchema == NULL || schemaVersion(*ppSchema) != TD_ROW_SVER(row)) {
*ppSchema = tsdbGetTableSchemaImpl(pTsdb, pTable, false, false, TD_ROW_SVER(row));
if (*ppSchema == NULL) {
ASSERT(false);
return -1;
} }
} }
} else {
px = pTbData->sl.pHead;
tdAppendSTSRowToDataCol(row, *ppSchema, pCols, merge); for (int8_t iLevel = pTbData->sl.maxLevel - 1; iLevel >= pTbData->sl.level; iLevel--) {
pos[iLevel] = px;
}
if (pTbData->sl.level) {
if (fromPos) px = pos[pTbData->sl.level - 1];
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_FORWARD(px, iLevel);
while (pn != pTbData->sl.pHead) {
pTKey = (TSDBKEY *)SL_NODE_DATA(pn);
c = tsdbKeyCmprFn(pTKey, pKey);
if (c >= 0) {
break;
} else {
px = pn;
pn = SL_NODE_FORWARD(px, iLevel);
}
}
pos[iLevel] = px;
}
}
}
}
static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
int8_t level = 1;
int8_t tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
const uint32_t factor = 4;
while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
level++;
} }
return 0; return level;
}
static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListNode **pos, TSDBROW *pRow,
int8_t forward) {
int32_t code = 0;
int8_t level;
SMemSkipListNode *pNode;
SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
// node
level = tsdbMemSkipListRandLevel(&pTbData->sl);
pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level) + tPutTSDBRow(NULL, pRow));
if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pNode->level = level;
for (int8_t iLevel = 0; iLevel < level; iLevel++) {
SL_NODE_FORWARD(pNode, iLevel) = NULL;
SL_NODE_BACKWARD(pNode, iLevel) = NULL;
}
tPutTSDBRow((uint8_t *)SL_NODE_DATA(pNode), pRow);
// put
for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
SMemSkipListNode *px = pos[iLevel];
if (forward) {
SMemSkipListNode *pNext = SL_NODE_FORWARD(px, iLevel);
SL_NODE_FORWARD(pNode, iLevel) = pNext;
SL_NODE_BACKWARD(pNode, iLevel) = px;
SL_NODE_BACKWARD(pNext, iLevel) = pNode;
SL_NODE_FORWARD(px, iLevel) = pNode;
} else {
SMemSkipListNode *pPrev = SL_NODE_BACKWARD(px, iLevel);
SL_NODE_FORWARD(pNode, iLevel) = px;
SL_NODE_BACKWARD(pNode, iLevel) = pPrev;
SL_NODE_FORWARD(pPrev, iLevel) = pNode;
SL_NODE_BACKWARD(px, iLevel) = pNode;
}
}
pTbData->sl.size++;
if (pTbData->sl.level < pNode->level) {
pTbData->sl.level = pNode->level;
}
_exit:
return code;
}
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) {
int32_t code = 0;
SSubmitBlkIter blkIter = {0};
TSDBKEY key = {.version = version};
SMemSkipListNode *pos[SL_MAX_LEVEL];
TSDBROW row = {.version = version, .pTSRow = NULL};
int32_t nRow = 0;
tInitSubmitBlkIter(pMsgIter, pBlock, &blkIter);
// backward put first data
row.pTSRow = tGetSubmitBlkNext(&blkIter);
key.ts = row.pTSRow->ts;
nRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_BACKWARD);
code = tbDataDoPut(pMemTable, pTbData, pos, &row, 0);
if (code) {
goto _err;
}
if (tsdbKeyCmprFn(&key, &pTbData->minKey) < 0) {
pTbData->minKey = key;
}
if (tsdbKeyCmprFn(&key, &pMemTable->minKey) < 0) {
pMemTable->minKey = key;
}
// forward put rest data
row.pTSRow = tGetSubmitBlkNext(&blkIter);
if (row.pTSRow) {
key.ts = row.pTSRow->ts;
for (int8_t iLevel = 0; iLevel < pTbData->sl.maxLevel; iLevel++) {
pos[iLevel] = SL_NODE_BACKWARD(pos[iLevel], iLevel);
}
do {
nRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
code = tbDataDoPut(pMemTable, pTbData, pos, &row, 1);
if (code) {
goto _err;
}
row.pTSRow = tGetSubmitBlkNext(&blkIter);
key.ts = row.pTSRow->ts;
} while (row.pTSRow);
}
if (tsdbKeyCmprFn(&key, &pTbData->maxKey) > 0) {
pTbData->maxKey = key;
}
if (tsdbKeyCmprFn(&key, &pMemTable->maxKey) > 0) {
pMemTable->maxKey = key;
}
pMemTable->nRef++;
pRsp->numOfRows = nRow;
pRsp->affectedRows = nRow;
return code;
_err:
return code;
}
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow) {
int32_t n = 0;
n += tPutI64(p, pRow->version);
if (p) memcpy(p + n, pRow->pTSRow, pRow->pTSRow->len);
n += pRow->pTSRow->len;
return n;
}
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow) {
int32_t n = 0;
n += tGetI64(p, &pRow->version);
pRow->pTSRow = (STSRow *)(p + n);
n += pRow->pTSRow->len;
return n;
} }

5
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
View File

@ -15,11 +15,6 @@
#include "tsdb.h" #include "tsdb.h"
struct SMemSkipListNode {
int8_t level;
SMemSkipListNode *forwards[0];
};
typedef struct { typedef struct {
tb_uid_t uid; tb_uid_t uid;
STSchema *pTSchema; STSchema *pTSchema;

124
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -67,15 +67,16 @@ enum {
}; };
typedef struct STableCheckInfo { typedef struct STableCheckInfo {
uint64_t tableId; uint64_t suid;
TSKEY lastKey; uint64_t tableId;
SBlockInfo* pCompInfo; TSKEY lastKey;
int32_t compSize; SBlockInfo* pCompInfo;
int32_t numOfBlocks : 29; // number of qualified data blocks not the original blocks int32_t compSize;
uint8_t chosen : 2; // indicate which iterator should move forward int32_t numOfBlocks : 29; // number of qualified data blocks not the original blocks
bool initBuf : 1; // whether to initialize the in-memory skip list iterator or not uint8_t chosen : 2; // indicate which iterator should move forward
SSkipListIterator* iter; // mem buffer skip list iterator bool initBuf : 1; // whether to initialize the in-memory skip list iterator or not
SSkipListIterator* iiter; // imem buffer skip list iterator STbDataIter* iter; // mem buffer skip list iterator
STbDataIter* iiter; // imem buffer skip list iterator
} STableCheckInfo; } STableCheckInfo;
typedef struct STableBlockInfo { typedef struct STableBlockInfo {
@ -265,8 +266,8 @@ static void resetCheckInfo(STsdbReadHandle* pTsdbReadHandle) {
for (int32_t i = 0; i < numOfTables; ++i) { for (int32_t i = 0; i < numOfTables; ++i) {
STableCheckInfo* pCheckInfo = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i); STableCheckInfo* pCheckInfo = (STableCheckInfo*)taosArrayGet(pTsdbReadHandle->pTableCheckInfo, i);
pCheckInfo->lastKey = pTsdbReadHandle->window.skey; pCheckInfo->lastKey = pTsdbReadHandle->window.skey;
pCheckInfo->iter = tSkipListDestroyIter(pCheckInfo->iter); pCheckInfo->iter = tsdbTbDataIterDestroy(pCheckInfo->iter);
pCheckInfo->iiter = tSkipListDestroyIter(pCheckInfo->iiter); pCheckInfo->iiter = tsdbTbDataIterDestroy(pCheckInfo->iiter);
pCheckInfo->initBuf = false; pCheckInfo->initBuf = false;
if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
@ -752,23 +753,21 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe
pCheckInfo->initBuf = true; pCheckInfo->initBuf = true;
int32_t order = pHandle->order; int32_t order = pHandle->order;
STbData** pMem = NULL; STbData* pMem = NULL;
STbData** pIMem = NULL; STbData* pIMem = NULL;
TSKEY tLastKey = keyToTkey(pCheckInfo->lastKey); TSKEY tLastKey = keyToTkey(pCheckInfo->lastKey);
if (pHandle->pTsdb->mem != NULL) { if (pHandle->pTsdb->mem != NULL) {
pMem = taosHashGet(pHandle->pTsdb->mem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId)); tsdbGetTbDataFromMemTable(pHandle->pTsdb->mem, pCheckInfo->suid, pCheckInfo->tableId, &pMem);
if (pMem != NULL) { if (pMem != NULL) {
pCheckInfo->iter = tsdbTbDataIterCreate(pMem, &(TSDBKEY){.version = 0, .ts = tLastKey}, 0, &pCheckInfo->iter);
tSkipListCreateIterFromVal((*pMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
} }
} }
if (pHandle->pTsdb->imem != NULL) { if (pHandle->pTsdb->imem != NULL) {
pIMem = taosHashGet(pHandle->pTsdb->imem->pHashIdx, &pCheckInfo->tableId, sizeof(pCheckInfo->tableId)); tsdbGetTbDataFromMemTable(pHandle->pTsdb->mem, pCheckInfo->suid, pCheckInfo->tableId, &pIMem);
if (pIMem != NULL) { if (pIMem != NULL) {
pCheckInfo->iiter = tsdbTbDataIterCreate(pIMem, &(TSDBKEY){.version = 0, .ts = tLastKey}, 0, &pCheckInfo->iiter);
tSkipListCreateIterFromVal((*pIMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
} }
} }
@ -777,22 +776,21 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe
return false; return false;
} }
bool memEmpty = (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tSkipListIterNext(pCheckInfo->iter)); bool memEmpty = (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tsdbTbDataIterNext(pCheckInfo->iter));
bool imemEmpty = (pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tSkipListIterNext(pCheckInfo->iiter)); bool imemEmpty = (pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tsdbTbDataIterNext(pCheckInfo->iiter));
if (memEmpty && imemEmpty) { // buffer is empty if (memEmpty && imemEmpty) { // buffer is empty
return false; return false;
} }
if (!memEmpty) { if (!memEmpty) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter); TSDBROW row;
assert(node != NULL);
STSRow* row = (STSRow*)SL_GET_NODE_DATA(node); tsdbTbDataIterGet(pCheckInfo->iter, &row);
TSKEY key = TD_ROW_KEY(row); // first timestamp in buffer TSKEY key = row.pTSRow->ts; // first timestamp in buffer
tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64 tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
"-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s", "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
pHandle, pCheckInfo->tableId, key, order, (*pMem)->minKey.ts, (*pMem)->maxKey.ts, pCheckInfo->lastKey, pHandle, pCheckInfo->tableId, key, order, pMem->minKey.ts, pMem->maxKey.ts, pCheckInfo->lastKey,
(*pMem)->nrows, pHandle->idStr); pMem->sl.size, pHandle->idStr);
if (ASCENDING_TRAVERSE(order)) { if (ASCENDING_TRAVERSE(order)) {
assert(pCheckInfo->lastKey <= key); assert(pCheckInfo->lastKey <= key);
@ -805,15 +803,14 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe
} }
if (!imemEmpty) { if (!imemEmpty) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter); TSDBROW row;
assert(node != NULL);
STSRow* row = (STSRow*)SL_GET_NODE_DATA(node); tsdbTbDataIterGet(pCheckInfo->iter, &row);
TSKEY key = TD_ROW_KEY(row); // first timestamp in buffer TSKEY key = row.pTSRow->ts; // first timestamp in buffer
tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64 tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
"-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s", "-%" PRId64 ", lastKey:%" PRId64 ", numOfRows:%" PRId64 ", %s",
pHandle, pCheckInfo->tableId, key, order, (*pIMem)->minKey.ts, (*pIMem)->maxKey.ts, pCheckInfo->lastKey, pHandle, pCheckInfo->tableId, key, order, pIMem->minKey.ts, pIMem->maxKey.ts, pCheckInfo->lastKey,
(*pIMem)->nrows, pHandle->idStr); pIMem->sl.size, pHandle->idStr);
if (ASCENDING_TRAVERSE(order)) { if (ASCENDING_TRAVERSE(order)) {
assert(pCheckInfo->lastKey <= key); assert(pCheckInfo->lastKey <= key);
@ -828,31 +825,23 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe
} }
static void destroyTableMemIterator(STableCheckInfo* pCheckInfo) { static void destroyTableMemIterator(STableCheckInfo* pCheckInfo) {
tSkipListDestroyIter(pCheckInfo->iter); tsdbTbDataIterDestroy(pCheckInfo->iter);
tSkipListDestroyIter(pCheckInfo->iiter); tsdbTbDataIterDestroy(pCheckInfo->iiter);
} }
static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, TDRowVerT maxVer) { static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, TDRowVerT maxVer) {
TSDBROW row = {0};
STSRow *rmem = NULL, *rimem = NULL; STSRow *rmem = NULL, *rimem = NULL;
if (pCheckInfo->iter) { if (pCheckInfo->iter) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter); if (tsdbTbDataIterGet(pCheckInfo->iter, &row)) {
if (node != NULL) { rmem = row.pTSRow;
rmem = (STSRow*)SL_GET_NODE_DATA(node);
// TODO: filter max version
// if (TD_ROW_VER(rmem) > maxVer) {
// rmem = NULL;
// }
} }
} }
if (pCheckInfo->iiter) { if (pCheckInfo->iiter) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter); if (tsdbTbDataIterGet(pCheckInfo->iiter, &row)) {
if (node != NULL) { rimem = row.pTSRow;
rimem = (STSRow*)SL_GET_NODE_DATA(node);
// TODO: filter max version
// if (TD_ROW_VER(rimem) > maxVer) {
// rimem = NULL;
// }
} }
} }
@ -889,7 +878,7 @@ static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order,
pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH; pCheckInfo->chosen = CHECKINFO_CHOSEN_BOTH;
} else { } else {
pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM; pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
tSkipListIterNext(pCheckInfo->iter); tsdbTbDataIterNext(pCheckInfo->iter);
} }
return r1; return r1;
} else if (r1 < r2 && ASCENDING_TRAVERSE(order)) { } else if (r1 < r2 && ASCENDING_TRAVERSE(order)) {
@ -903,28 +892,17 @@ static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order,
static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, STSRow** extraRow, static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int32_t update, STSRow** extraRow,
TDRowVerT maxVer) { TDRowVerT maxVer) {
TSDBROW row;
STSRow *rmem = NULL, *rimem = NULL; STSRow *rmem = NULL, *rimem = NULL;
if (pCheckInfo->iter) { if (pCheckInfo->iter) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iter); if (tsdbTbDataIterGet(pCheckInfo->iter, &row)) {
if (node != NULL) { rmem = row.pTSRow;
rmem = (STSRow*)SL_GET_NODE_DATA(node);
#if 0 // TODO: skiplist refactor
if (TD_ROW_VER(rmem) > maxVer) {
rmem = NULL;
}
#endif
} }
} }
if (pCheckInfo->iiter) { if (pCheckInfo->iiter) {
SSkipListNode* node = tSkipListIterGet(pCheckInfo->iiter); if (tsdbTbDataIterGet(pCheckInfo->iiter, &row)) {
if (node != NULL) { rimem = row.pTSRow;
rimem = (STSRow*)SL_GET_NODE_DATA(node);
#if 0 // TODO: skiplist refactor
if (TD_ROW_VER(rimem) > maxVer) {
rimem = NULL;
}
#endif
} }
} }
@ -966,7 +944,7 @@ static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int
*extraRow = rimem; *extraRow = rimem;
return rmem; return rmem;
} else { } else {
tSkipListIterNext(pCheckInfo->iter); tsdbTbDataIterNext(pCheckInfo->iter);
pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM; pCheckInfo->chosen = CHECKINFO_CHOSEN_IMEM;
return rimem; return rimem;
} }
@ -995,7 +973,7 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) {
bool hasNext = false; bool hasNext = false;
if (pCheckInfo->chosen == CHECKINFO_CHOSEN_MEM) { if (pCheckInfo->chosen == CHECKINFO_CHOSEN_MEM) {
if (pCheckInfo->iter != NULL) { if (pCheckInfo->iter != NULL) {
hasNext = tSkipListIterNext(pCheckInfo->iter); hasNext = tsdbTbDataIterNext(pCheckInfo->iter);
} }
if (hasNext) { if (hasNext) {
@ -1003,11 +981,11 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) {
} }
if (pCheckInfo->iiter != NULL) { if (pCheckInfo->iiter != NULL) {
return tSkipListIterGet(pCheckInfo->iiter) != NULL; return tsdbTbDataIterGet(pCheckInfo->iiter, NULL);
} }
} else if (pCheckInfo->chosen == CHECKINFO_CHOSEN_IMEM) { } else if (pCheckInfo->chosen == CHECKINFO_CHOSEN_IMEM) {
if (pCheckInfo->iiter != NULL) { if (pCheckInfo->iiter != NULL) {
hasNext = tSkipListIterNext(pCheckInfo->iiter); hasNext = tsdbTbDataIterNext(pCheckInfo->iiter);
} }
if (hasNext) { if (hasNext) {
@ -1015,14 +993,14 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) {
} }
if (pCheckInfo->iter != NULL) { if (pCheckInfo->iter != NULL) {
return tSkipListIterGet(pCheckInfo->iter) != NULL; return tsdbTbDataIterGet(pCheckInfo->iter, NULL);
} }
} else { } else {
if (pCheckInfo->iter != NULL) { if (pCheckInfo->iter != NULL) {
hasNext = tSkipListIterNext(pCheckInfo->iter); hasNext = tsdbTbDataIterNext(pCheckInfo->iter);
} }
if (pCheckInfo->iiter != NULL) { if (pCheckInfo->iiter != NULL) {
hasNext = tSkipListIterNext(pCheckInfo->iiter) || hasNext; hasNext = tsdbTbDataIterNext(pCheckInfo->iiter) || hasNext;
} }
} }

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

@ -39,7 +39,7 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
SSubmitBlkRsp r = {0}; SSubmitBlkRsp r = {0};
tGetSubmitMsgNext(&msgIter, &pBlock); tGetSubmitMsgNext(&msgIter, &pBlock);
if (pBlock == NULL) break; if (pBlock == NULL) break;
if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &r) < 0) { if (tsdbInsertTableData(pTsdb, version, &msgIter, pBlock, &r) < 0) {
return -1; return -1;
} }

2
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -779,7 +779,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname); sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname);
} }
if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &submitBlkRsp) < 0) { if (tsdbInsertTableData(pVnode->pTsdb, version, &msgIter, pBlock, &submitBlkRsp) < 0) {
submitBlkRsp.code = terrno; submitBlkRsp.code = terrno;
} }