Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 84aac9e3 编写于 作者: H Hongze Cheng 提交者： GitHub
浏览文件
操作

Merge pull request #13563 from taosdata/feat/row_refact 

feat: vnode multi version
上级 58ff4429 fa8c6f26
隐藏空白更改
内联 并排
Showing with 911 addition and 530 deletion
include/common/tcommon.h
浏览文件 @ 84aac9e3
...@@ -108,6 +108,7 @@ typedef struct SColumnInfoData { ...@@ -108,6 +108,7 @@ typedef struct SColumnInfoData {
typedef struct SQueryTableDataCond { typedef struct SQueryTableDataCond {
// STimeWindow twindow; // STimeWindow twindow;
uint64_t suid;
int32_t order; // desc|asc order to iterate the data block int32_t order; // desc|asc order to iterate the data block
int32_t numOfCols; int32_t numOfCols;
SColumnInfo* colList; SColumnInfo* colList;
......
source/dnode/vnode/CMakeLists.txt
浏览文件 @ 84aac9e3
...@@ -36,12 +36,12 @@ target_sources( ...@@ -36,12 +36,12 @@ target_sources(
# tsdb # tsdb
"src/tsdb/tsdbCommit.c" "src/tsdb/tsdbCommit.c"
"src/tsdb/tsdbCommit2.c" # "src/tsdb/tsdbCommit2.c"
"src/tsdb/tsdbFile.c" "src/tsdb/tsdbFile.c"
"src/tsdb/tsdbFS.c" "src/tsdb/tsdbFS.c"
"src/tsdb/tsdbOpen.c" "src/tsdb/tsdbOpen.c"
"src/tsdb/tsdbMemTable.c" "src/tsdb/tsdbMemTable.c"
"src/tsdb/tsdbMemTable2.c" # "src/tsdb/tsdbMemTable2.c"
"src/tsdb/tsdbRead.c" "src/tsdb/tsdbRead.c"
"src/tsdb/tsdbReadImpl.c" "src/tsdb/tsdbReadImpl.c"
"src/tsdb/tsdbWrite.c" "src/tsdb/tsdbWrite.c"
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 84aac9e3
...@@ -39,24 +39,46 @@ typedef struct SDelOp SDelOp; ...@@ -39,24 +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 ==============================================================================================
typedef struct STbData STbData;
typedef struct SMemTable SMemTable;
typedef struct STbDataIter STbDataIter;
typedef struct SMergeInfo SMergeInfo;
typedef struct STable STable;
// SMemTable
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
void tsdbMemTableDestroy(SMemTable *pMemTable);
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, STbDataIter *pIter, TSKEY maxKey, int maxRowsToRead,
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
// tsdbMemTable2.c ============================================================================================== // tsdbMemTable2.c ==============================================================================================
typedef struct SMemTable SMemTable; // 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, SMemTable **ppMemTable); // int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable2 **ppMemTable);
void tsdbMemTableDestroy2(SMemTable *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;
...@@ -124,17 +146,6 @@ int tsdbRLockFS(STsdbFS *pFs); ...@@ -124,17 +146,6 @@ int tsdbRLockFS(STsdbFS *pFs);
int tsdbWLockFS(STsdbFS *pFs); int tsdbWLockFS(STsdbFS *pFs);
int tsdbUnLockFS(STsdbFS *pFs); int tsdbUnLockFS(STsdbFS *pFs);
// tsdbMemTable ================
typedef struct STbData STbData;
typedef struct STsdbMemTable STsdbMemTable;
typedef struct SMergeInfo SMergeInfo;
typedef struct STable STable;
int tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
void tsdbMemTableDestroy(STsdbMemTable *pMemTable);
int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
// structs // structs
typedef struct { typedef struct {
int minFid; int minFid;
...@@ -145,30 +156,27 @@ typedef struct { ...@@ -145,30 +156,27 @@ typedef struct {
#define TSDB_DATA_DIR_LEN 6 // adapt accordingly #define TSDB_DATA_DIR_LEN 6 // adapt accordingly
struct STsdb { struct STsdb {
char *path; char *path;
SVnode *pVnode; SVnode *pVnode;
TdThreadMutex mutex; TdThreadMutex mutex;
char dir[TSDB_DATA_DIR_LEN]; char dir[TSDB_DATA_DIR_LEN];
bool repoLocked; bool repoLocked;
STsdbKeepCfg keepCfg; STsdbKeepCfg keepCfg;
STsdbMemTable *mem; SMemTable *mem;
STsdbMemTable *imem; SMemTable *imem;
SRtn rtn; SRtn rtn;
STsdbFS *fs; STsdbFS *fs;
}; };
#if 1 // ====================================== #if 1 // ======================================
struct STable { struct STable {
uint64_t tid; uint64_t suid;
uint64_t uid; uint64_t uid;
STSchema *pSchema; // latest schema STSchema *pSchema; // latest schema
STSchema *pCacheSchema; // cached cache STSchema *pCacheSchema; // cached cache
}; };
#define TABLE_TID(t) (t)->tid
#define TABLE_UID(t) (t)->uid
// int tsdbPrepareCommit(STsdb *pTsdb); // int tsdbPrepareCommit(STsdb *pTsdb);
typedef enum { typedef enum {
TSDB_FILE_HEAD = 0, // .head TSDB_FILE_HEAD = 0, // .head
...@@ -206,27 +214,44 @@ struct SDFileSet { ...@@ -206,27 +214,44 @@ struct SDFileSet {
SDFile files[TSDB_FILE_MAX]; SDFile files[TSDB_FILE_MAX];
}; };
struct TSDBKEY {
int64_t version;
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 uid; tb_uid_t suid;
TSKEY keyMin; tb_uid_t uid;
TSKEY keyMax; TSDBKEY minKey;
int64_t minVer; TSDBKEY maxKey;
int64_t maxVer; SDelOp *pHead;
int64_t nrows; SDelOp *pTail;
SSkipList *pData; SMemSkipList sl;
}; };
struct STsdbMemTable { struct SMemTable {
SVBufPool *pPool; SRWLatch latch;
T_REF_DECLARE() STsdb *pTsdb;
SRWLatch latch; int32_t nRef;
TSKEY keyMin; TSDBKEY minKey;
TSKEY keyMax; TSDBKEY maxKey;
int64_t minVer; int64_t nRow;
int64_t maxVer; int64_t nDelOp;
int64_t nRow; SArray *aTbData; // SArray<STbData>
SSkipList *pSlIdx; // SSkiplist<STbData>
SHashObj *pHashIdx;
}; };
struct STsdbFSMeta { struct STsdbFSMeta {
...@@ -237,9 +262,11 @@ struct STsdbFSMeta { ...@@ -237,9 +262,11 @@ struct STsdbFSMeta {
// ================== // ==================
typedef struct { typedef struct {
STsdbFSMeta meta; // FS meta STsdbFSMeta meta; // FS meta
SArray *df; // data file array SDFile cacheFile; // cache file
SArray *sf; // sma data file array v2f1900.index_name_1 SDFile tombstone; // tomestome file
SArray *df; // data file array
SArray *sf; // sma data file array v2f1900.index_name_1
} SFSStatus; } SFSStatus;
struct STsdbFS { struct STsdbFS {
...@@ -292,16 +319,24 @@ static void *taosTZfree(void *ptr); ...@@ -292,16 +319,24 @@ static void *taosTZfree(void *ptr);
static size_t taosTSizeof(void *ptr); static size_t taosTSizeof(void *ptr);
static void taosTMemset(void *ptr, int c); static void taosTMemset(void *ptr, int c);
static FORCE_INLINE STSRow *tsdbNextIterRow(SSkipListIterator *pIter) { struct TSDBROW {
int64_t version;
STSRow *pTSRow;
};
static FORCE_INLINE STSRow *tsdbNextIterRow(STbDataIter *pIter) {
TSDBROW row;
if (pIter == NULL) return NULL; if (pIter == NULL) return NULL;
SSkipListNode *node = tSkipListIterGet(pIter); if (tsdbTbDataIterGet(pIter, &row)) {
if (node == NULL) return NULL; return row.pTSRow;
}
return (STSRow *)SL_GET_NODE_DATA(node); return NULL;
} }
static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator *pIter) { static FORCE_INLINE TSKEY tsdbNextIterKey(STbDataIter *pIter) {
STSRow *row = tsdbNextIterRow(pIter); STSRow *row = tsdbNextIterRow(pIter);
if (row == NULL) return TSDB_DATA_TIMESTAMP_NULL; if (row == NULL) return TSDB_DATA_TIMESTAMP_NULL;
...@@ -311,11 +346,6 @@ static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator *pIter) { ...@@ -311,11 +346,6 @@ static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator *pIter) {
// tsdbReadImpl // tsdbReadImpl
typedef struct SReadH SReadH; typedef struct SReadH SReadH;
struct TSDBKEY {
int64_t version;
TSKEY ts;
};
typedef struct { typedef struct {
uint64_t suid; uint64_t suid;
uint64_t uid; uint64_t uid;
...@@ -354,7 +384,7 @@ typedef struct { ...@@ -354,7 +384,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t delimiter; // For recovery usage int32_t delimiter; // For recovery usage
int32_t tid; uint64_t suid;
uint64_t uid; uint64_t uid;
SBlock blocks[]; SBlock blocks[];
} SBlockInfo; } SBlockInfo;
...@@ -650,11 +680,6 @@ struct SFSIter { ...@@ -650,11 +680,6 @@ struct SFSIter {
#define TSDB_FS_ITER_FORWARD TSDB_ORDER_ASC #define TSDB_FS_ITER_FORWARD TSDB_ORDER_ASC
#define TSDB_FS_ITER_BACKWARD TSDB_ORDER_DESC #define TSDB_FS_ITER_BACKWARD TSDB_ORDER_DESC
struct TSDBROW {
int64_t version;
STSRow2 tsRow;
};
struct TABLEID { struct TABLEID {
tb_uid_t suid; tb_uid_t suid;
tb_uid_t uid; tb_uid_t uid;
...@@ -675,7 +700,7 @@ typedef struct { ...@@ -675,7 +700,7 @@ typedef struct {
TSKEY eKey; TSKEY eKey;
} SDelInfo; } SDelInfo;
struct SMemTable { struct SMemTable2 {
STsdb *pTsdb; STsdb *pTsdb;
int32_t nRef; int32_t nRef;
TSDBKEY minKey; TSDBKEY minKey;
...@@ -705,16 +730,6 @@ static FORCE_INLINE int tsdbKeyCmprFn(const void *p1, const void *p2) { ...@@ -705,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;
...@@ -726,13 +741,19 @@ struct SMemData { ...@@ -726,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
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 84aac9e3
...@@ -116,7 +116,9 @@ int tsdbBegin(STsdb* pTsdb); ...@@ -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,
......
source/dnode/vnode/src/meta/metaQuery.c
浏览文件 @ 84aac9e3
...@@ -31,7 +31,7 @@ void metaReaderClear(SMetaReader *pReader) { ...@@ -31,7 +31,7 @@ void metaReaderClear(SMetaReader *pReader) {
} }
int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t uid) { int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t uid) {
SMeta * pMeta = pReader->pMeta; SMeta *pMeta = pReader->pMeta;
STbDbKey tbDbKey = {.version = version, .uid = uid}; STbDbKey tbDbKey = {.version = version, .uid = uid};
// query table.db // query table.db
...@@ -54,7 +54,7 @@ _err: ...@@ -54,7 +54,7 @@ _err:
} }
int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) { int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
SMeta * pMeta = pReader->pMeta; SMeta *pMeta = pReader->pMeta;
int64_t version; int64_t version;
// query uid.idx // query uid.idx
...@@ -68,7 +68,7 @@ int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) { ...@@ -68,7 +68,7 @@ int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
} }
int metaGetTableEntryByName(SMetaReader *pReader, const char *name) { int metaGetTableEntryByName(SMetaReader *pReader, const char *name) {
SMeta * pMeta = pReader->pMeta; SMeta *pMeta = pReader->pMeta;
tb_uid_t uid; tb_uid_t uid;
// query name.idx // query name.idx
...@@ -82,7 +82,7 @@ int metaGetTableEntryByName(SMetaReader *pReader, const char *name) { ...@@ -82,7 +82,7 @@ int metaGetTableEntryByName(SMetaReader *pReader, const char *name) {
} }
tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) { tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) {
void * pData = NULL; void *pData = NULL;
int nData = 0; int nData = 0;
tb_uid_t uid = 0; tb_uid_t uid = 0;
...@@ -134,7 +134,7 @@ void metaCloseTbCursor(SMTbCursor *pTbCur) { ...@@ -134,7 +134,7 @@ void metaCloseTbCursor(SMTbCursor *pTbCur) {
int metaTbCursorNext(SMTbCursor *pTbCur) { int metaTbCursorNext(SMTbCursor *pTbCur) {
int ret; int ret;
void * pBuf; void *pBuf;
STbCfg tbCfg; STbCfg tbCfg;
for (;;) { for (;;) {
...@@ -155,7 +155,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur) { ...@@ -155,7 +155,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur) {
} }
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) { SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
void * pData = NULL; void *pData = NULL;
int nData = 0; int nData = 0;
int64_t version; int64_t version;
SSchemaWrapper schema = {0}; SSchemaWrapper schema = {0};
...@@ -163,37 +163,47 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo ...@@ -163,37 +163,47 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
SDecoder dc = {0}; SDecoder dc = {0};
metaRLock(pMeta); metaRLock(pMeta);
if (sver < 0) { _query:
if (tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), &pData, &nData) < 0) { if (tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), &pData, &nData) < 0) {
goto _err; goto _err;
} }
version = *(int64_t *)pData;
tdbTbGet(pMeta->pTbDb, &(STbDbKey){.uid = uid, .version = version}, sizeof(STbDbKey), &pData, &nData); version = *(int64_t *)pData;
SMetaEntry me = {0}; tdbTbGet(pMeta->pTbDb, &(STbDbKey){.uid = uid, .version = version}, sizeof(STbDbKey), &pData, &nData);
tDecoderInit(&dc, pData, nData); SMetaEntry me = {0};
metaDecodeEntry(&dc, &me); tDecoderInit(&dc, pData, nData);
if (me.type == TSDB_SUPER_TABLE) { metaDecodeEntry(&dc, &me);
if (me.type == TSDB_SUPER_TABLE) {
if (sver == -1 || sver == me.stbEntry.schemaRow.version) {
pSchema = tCloneSSchemaWrapper(&me.stbEntry.schemaRow); pSchema = tCloneSSchemaWrapper(&me.stbEntry.schemaRow);
} else if (me.type == TSDB_NORMAL_TABLE) { tDecoderClear(&dc);
pSchema = tCloneSSchemaWrapper(&me.ntbEntry.schemaRow); goto _exit;
} else {
ASSERT(0);
} }
} else if (me.type == TSDB_CHILD_TABLE) {
uid = me.ctbEntry.suid;
tDecoderClear(&dc); tDecoderClear(&dc);
goto _query;
} else { } else {
if (tdbTbGet(pMeta->pSkmDb, &(SSkmDbKey){.uid = uid, .sver = sver}, sizeof(SSkmDbKey), &pData, &nData) < 0) { if (sver == -1 || sver == me.ntbEntry.schemaRow.version) {
goto _err; pSchema = tCloneSSchemaWrapper(&me.ntbEntry.schemaRow);
tDecoderClear(&dc);
goto _exit;
} }
}
tDecoderClear(&dc);
tDecoderInit(&dc, pData, nData); // query from skm db
tDecodeSSchemaWrapper(&dc, &schema); if (tdbTbGet(pMeta->pSkmDb, &(SSkmDbKey){.uid = uid, .sver = sver}, sizeof(SSkmDbKey), &pData, &nData) < 0) {
pSchema = tCloneSSchemaWrapper(&schema); goto _err;
tDecoderClear(&dc);
} }
tDecoderInit(&dc, pData, nData);
tDecodeSSchemaWrapper(&dc, &schema);
pSchema = tCloneSSchemaWrapper(&schema);
tDecoderClear(&dc);
_exit:
metaULock(pMeta); metaULock(pMeta);
tdbFree(pData); tdbFree(pData);
return pSchema; return pSchema;
...@@ -205,11 +215,11 @@ _err: ...@@ -205,11 +215,11 @@ _err:
} }
struct SMCtbCursor { struct SMCtbCursor {
SMeta * pMeta; SMeta *pMeta;
TBC * pCur; TBC *pCur;
tb_uid_t suid; tb_uid_t suid;
void * pKey; void *pKey;
void * pVal; void *pVal;
int kLen; int kLen;
int vLen; int vLen;
}; };
...@@ -279,25 +289,13 @@ tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) { ...@@ -279,25 +289,13 @@ tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
} }
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) { STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
tb_uid_t quid; // SMetaReader mr = {0};
SMetaReader mr = {0}; STSchema *pTSchema = NULL;
STSchema * pTSchema = NULL;
SSchemaWrapper *pSW = NULL; SSchemaWrapper *pSW = NULL;
STSchemaBuilder sb = {0}; STSchemaBuilder sb = {0};
SSchema * pSchema; SSchema *pSchema;
metaReaderInit(&mr, pMeta, 0);
metaGetTableEntryByUid(&mr, uid);
if (mr.me.type == TSDB_CHILD_TABLE) {
quid = mr.me.ctbEntry.suid;
} else {
quid = uid;
}
metaReaderClear(&mr);
pSW = metaGetTableSchema(pMeta, quid, sver, 0); pSW = metaGetTableSchema(pMeta, uid, sver, 0);
if (!pSW) return NULL; if (!pSW) return NULL;
tdInitTSchemaBuilder(&sb, pSW->version); tdInitTSchemaBuilder(&sb, pSW->version);
...@@ -321,11 +319,11 @@ int metaGetTbNum(SMeta *pMeta) { ...@@ -321,11 +319,11 @@ int metaGetTbNum(SMeta *pMeta) {
} }
typedef struct { typedef struct {
SMeta * pMeta; SMeta *pMeta;
TBC * pCur; TBC *pCur;
tb_uid_t uid; tb_uid_t uid;
void * pKey; void *pKey;
void * pVal; void *pVal;
int kLen; int kLen;
int vLen; int vLen;
} SMSmaCursor; } SMSmaCursor;
...@@ -397,7 +395,7 @@ tb_uid_t metaSmaCursorNext(SMSmaCursor *pSmaCur) { ...@@ -397,7 +395,7 @@ tb_uid_t metaSmaCursorNext(SMSmaCursor *pSmaCur) {
STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) { STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {
STSmaWrapper *pSW = NULL; STSmaWrapper *pSW = NULL;
SArray * pSmaIds = NULL; SArray *pSmaIds = NULL;
if (!(pSmaIds = metaGetSmaIdsByTable(pMeta, uid))) { if (!(pSmaIds = metaGetSmaIdsByTable(pMeta, uid))) {
return NULL; return NULL;
...@@ -421,7 +419,7 @@ STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) { ...@@ -421,7 +419,7 @@ STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {
metaReaderInit(&mr, pMeta, 0); metaReaderInit(&mr, pMeta, 0);
int64_t smaId; int64_t smaId;
int smaIdx = 0; int smaIdx = 0;
STSma * pTSma = NULL; STSma *pTSma = NULL;
for (int i = 0; i < pSW->number; ++i) { for (int i = 0; i < pSW->number; ++i) {
smaId = *(tb_uid_t *)taosArrayGet(pSmaIds, i); smaId = *(tb_uid_t *)taosArrayGet(pSmaIds, i);
if (metaGetTableEntryByUid(&mr, smaId) < 0) { if (metaGetTableEntryByUid(&mr, smaId) < 0) {
...@@ -469,7 +467,7 @@ _err: ...@@ -469,7 +467,7 @@ _err:
} }
STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) { STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
STSma * pTSma = NULL; STSma *pTSma = NULL;
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, pMeta, 0); metaReaderInit(&mr, pMeta, 0);
if (metaGetTableEntryByUid(&mr, indexUid) < 0) { if (metaGetTableEntryByUid(&mr, indexUid) < 0) {
...@@ -491,7 +489,7 @@ STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) { ...@@ -491,7 +489,7 @@ STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
} }
SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) { SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {
SArray * pUids = NULL; SArray *pUids = NULL;
SSmaIdxKey *pSmaIdxKey = NULL; SSmaIdxKey *pSmaIdxKey = NULL;
SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid); SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid);
...@@ -529,7 +527,7 @@ SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) { ...@@ -529,7 +527,7 @@ SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {
} }
SArray *metaGetSmaTbUids(SMeta *pMeta) { SArray *metaGetSmaTbUids(SMeta *pMeta) {
SArray * pUids = NULL; SArray *pUids = NULL;
SSmaIdxKey *pSmaIdxKey = NULL; SSmaIdxKey *pSmaIdxKey = NULL;
tb_uid_t lastUid = 0; tb_uid_t lastUid = 0;
...@@ -591,13 +589,13 @@ const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) { ...@@ -591,13 +589,13 @@ const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) {
} }
typedef struct { typedef struct {
SMeta * pMeta; SMeta *pMeta;
TBC * pCur; TBC *pCur;
tb_uid_t suid; tb_uid_t suid;
int16_t cid; int16_t cid;
int16_t type; int16_t type;
void * pKey; void *pKey;
void * pVal; void *pVal;
int32_t kLen; int32_t kLen;
int32_t vLen; int32_t vLen;
} SIdxCursor; } SIdxCursor;
...@@ -621,7 +619,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { ...@@ -621,7 +619,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t nKey = 0; int32_t nKey = 0;
int32_t nTagData = 0; int32_t nTagData = 0;
void * tagData = NULL; void *tagData = NULL;
if (IS_VAR_DATA_TYPE(param->type)) { if (IS_VAR_DATA_TYPE(param->type)) {
tagData = varDataVal(param->val); tagData = varDataVal(param->val);
...@@ -640,7 +638,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { ...@@ -640,7 +638,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
goto END; goto END;
} }
void * entryKey = NULL, *entryVal = NULL; void *entryKey = NULL, *entryVal = NULL;
int32_t nEntryKey, nEntryVal; int32_t nEntryKey, nEntryVal;
while (1) { while (1) {
valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal); valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal);
......
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 84aac9e3
...@@ -141,10 +141,10 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p ...@@ -141,10 +141,10 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
} }
// this interface use suid instead of uid // this interface use suid instead of uid
pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true); pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion, true);
if (pHandle->pSchemaWrapper == NULL) { if (pHandle->pSchemaWrapper == NULL) {
tqWarn("cannot found schema wrapper for table: suid: %ld, version %d, possibly dropped table", tqWarn("cannot found schema wrapper for table: suid: %ld, version %d, possibly dropped table",
pHandle->msgIter.suid, pHandle->cachedSchemaVer); pHandle->msgIter.uid, pHandle->cachedSchemaVer);
/*ASSERT(0);*/ /*ASSERT(0);*/
terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND; terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
return -1; return -1;
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 84aac9e3
...@@ -18,8 +18,8 @@ ...@@ -18,8 +18,8 @@
#define TSDB_MAX_SUBBLOCKS 8 #define TSDB_MAX_SUBBLOCKS 8
typedef struct { typedef struct {
STable *pTable; STable *pTable;
SSkipListIterator *pIter; STbDataIter *pIter;
} SCommitIter; } SCommitIter;
typedef struct { typedef struct {
...@@ -58,26 +58,26 @@ typedef struct { ...@@ -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);
...@@ -92,7 +92,7 @@ static int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn); ...@@ -92,7 +92,7 @@ static int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn);
int tsdbBegin(STsdb *pTsdb) { int tsdbBegin(STsdb *pTsdb) {
if (!pTsdb) return 0; if (!pTsdb) return 0;
STsdbMemTable *pMem; SMemTable *pMem;
if (tsdbMemTableCreate(pTsdb, &pTsdb->mem) < 0) { if (tsdbMemTableCreate(pTsdb, &pTsdb->mem) < 0) {
return -1; return -1;
...@@ -244,7 +244,7 @@ void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) { ...@@ -244,7 +244,7 @@ void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
} }
static void tsdbStartCommit(STsdb *pRepo) { static void tsdbStartCommit(STsdb *pRepo) {
STsdbMemTable *pMem = pRepo->imem; SMemTable *pMem = pRepo->imem;
tsdbInfo("vgId:%d, start to commit", REPO_ID(pRepo)); tsdbInfo("vgId:%d, start to commit", REPO_ID(pRepo));
...@@ -400,7 +400,7 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) { ...@@ -400,7 +400,7 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
break; break;
} }
if (pIter && pIter->pTable && (!pIdx || (pIter->pTable->uid <= pIdx->uid))) { if (pIter && pIter->pTable && (!pIdx || (pIter->pTable->suid <= pIdx->suid || pIter->pTable->uid <= pIdx->uid))) {
if (tsdbCommitToTable(pCommith, mIter) < 0) { if (tsdbCommitToTable(pCommith, mIter) < 0) {
tsdbCloseCommitFile(pCommith, true); tsdbCloseCommitFile(pCommith, true);
// revert the file change // revert the file change
...@@ -453,57 +453,48 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) { ...@@ -453,57 +453,48 @@ 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;
STsdbMemTable *pMem = pRepo->imem; STsdb *pRepo = TSDB_COMMIT_REPO(pCommith);
SSkipListIterator *pSlIter; SMemTable *pMem = pRepo->imem;
SCommitIter *pCommitIter; STbData *pTbData;
SSkipListNode *pNode; SCommitIter *pCommitIter;
STbData *pTbData; STSchema *pTSchema = NULL;
STSchema *pTSchema = NULL;
pCommith->niters = SL_SIZE(pMem->pSlIdx); pCommith->niters = taosArrayGetSize(pMem->aTbData);
pCommith->iters = (SCommitIter *)taosMemoryCalloc(pCommith->niters, sizeof(SCommitIter)); pCommith->iters = (SCommitIter *)taosMemoryCalloc(pCommith->niters, sizeof(SCommitIter));
if (pCommith->iters == NULL) { if (pCommith->iters == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
return -1; goto _err;
} }
// Loop to create iters for each skiplist for (int32_t iIter = 0; iIter < pCommith->niters; iIter++) {
pSlIter = tSkipListCreateIter(pMem->pSlIdx); pTbData = (STbData *)taosArrayGetP(pMem->aTbData, iIter);
if (pSlIter == NULL) { pCommitIter = &pCommith->iters[iIter];
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
for (int i = 0; i < pCommith->niters; i++) {
tSkipListIterNext(pSlIter);
pNode = tSkipListIterGet(pSlIter);
pTbData = (STbData *)pNode->pData;
pCommitIter = pCommith->iters + i;
pTSchema = metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, -1); pTSchema = metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, -1);
if (pTSchema) { if (pTSchema) {
pCommitIter->pIter = tSkipListCreateIter(pTbData->pData); tsdbTbDataIterCreate(pTbData, NULL, 0, &pCommitIter->pIter);
tSkipListIterNext(pCommitIter->pIter);
pCommitIter->pTable = (STable *)taosMemoryMalloc(sizeof(STable)); pCommitIter->pTable = (STable *)taosMemoryMalloc(sizeof(STable));
pCommitIter->pTable->uid = pTbData->uid; pCommitIter->pTable->uid = pTbData->uid;
pCommitIter->pTable->tid = pTbData->uid; pCommitIter->pTable->suid = pTbData->suid;
pCommitIter->pTable->pSchema = pTSchema; pCommitIter->pTable->pSchema = pTSchema;
pCommitIter->pTable->pCacheSchema = NULL; pCommitIter->pTable->pCacheSchema = NULL;
} }
} }
tSkipListDestroyIter(pSlIter);
return 0; return code;
_err:
return code;
} }
static void tsdbDestroyCommitIters(SCommitH *pCommith) { static void tsdbDestroyCommitIters(SCommitH *pCommith) {
if (pCommith->iters == NULL) return; if (pCommith->iters == NULL) return;
for (int i = 1; i < pCommith->niters; i++) { for (int i = 1; i < pCommith->niters; i++) {
tSkipListDestroyIter(pCommith->iters[i].pIter); tsdbTbDataIterDestroy(pCommith->iters[i].pIter);
if (pCommith->iters[i].pTable) { if (pCommith->iters[i].pTable) {
tdFreeSchema(pCommith->iters[i].pTable->pSchema); tdFreeSchema(pCommith->iters[i].pTable->pSchema);
tdFreeSchema(pCommith->iters[i].pTable->pCacheSchema); tdFreeSchema(pCommith->iters[i].pTable->pCacheSchema);
...@@ -743,8 +734,8 @@ static int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, ...@@ -743,8 +734,8 @@ static int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA,
pBlkInfo = *ppBuf; pBlkInfo = *ppBuf;
pBlkInfo->delimiter = TSDB_FILE_DELIMITER; pBlkInfo->delimiter = TSDB_FILE_DELIMITER;
pBlkInfo->tid = TABLE_TID(pTable); pBlkInfo->suid = pTable->suid;
pBlkInfo->uid = TABLE_UID(pTable); pBlkInfo->uid = pTable->uid;
memcpy((void *)(pBlkInfo->blocks), taosArrayGet(pSupA, 0), nSupBlocks * sizeof(SBlock)); memcpy((void *)(pBlkInfo->blocks), taosArrayGet(pSupA, 0), nSupBlocks * sizeof(SBlock));
if (nSubBlocks > 0) { if (nSubBlocks > 0) {
...@@ -770,7 +761,8 @@ static int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, ...@@ -770,7 +761,8 @@ static int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA,
// Set pIdx // Set pIdx
pBlock = taosArrayGetLast(pSupA); pBlock = taosArrayGetLast(pSupA);
pIdx->uid = TABLE_UID(pTable); pIdx->suid = pTable->suid;
pIdx->uid = pTable->uid;
pIdx->hasLast = pBlock->last ? 1 : 0; pIdx->hasLast = pBlock->last ? 1 : 0;
pIdx->maxKey = pBlock->maxKey; pIdx->maxKey = pBlock->maxKey;
pIdx->numOfBlocks = (uint32_t)nSupBlocks; pIdx->numOfBlocks = (uint32_t)nSupBlocks;
...@@ -925,7 +917,7 @@ static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) { ...@@ -925,7 +917,7 @@ static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) {
return -1; return -1;
} }
STable table = {.tid = pIdx->uid, .uid = pIdx->uid, .pSchema = NULL}; STable table = {.suid = pIdx->suid, .uid = pIdx->uid, .pSchema = NULL};
pCommith->pTable = &table; pCommith->pTable = &table;
while (bidx < nBlocks) { while (bidx < nBlocks) {
...@@ -1186,7 +1178,7 @@ static int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFi ...@@ -1186,7 +1178,7 @@ static int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFi
} }
pBlockData->delimiter = TSDB_FILE_DELIMITER; pBlockData->delimiter = TSDB_FILE_DELIMITER;
pBlockData->uid = TABLE_UID(pTable); pBlockData->uid = pTable->uid;
pBlockData->numOfCols = nColsNotAllNull; pBlockData->numOfCols = nColsNotAllNull;
taosCalcChecksumAppend(0, (uint8_t *)pBlockData, tsize); taosCalcChecksumAppend(0, (uint8_t *)pBlockData, tsize);
...@@ -1226,7 +1218,7 @@ static int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFi ...@@ -1226,7 +1218,7 @@ static int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFi
tsdbDebug("vgId:%d, uid:%" PRId64 " a block of data is written to file %s, offset %" PRId64 tsdbDebug("vgId:%d, uid:%" PRId64 " a block of data is written to file %s, offset %" PRId64
" numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64, " numOfRows %d len %d numOfCols %" PRId16 " keyFirst %" PRId64 " keyLast %" PRId64,
REPO_ID(pRepo), TABLE_UID(pTable), TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len, REPO_ID(pRepo), pTable->uid, TSDB_FILE_FULL_NAME(pDFile), offset, rowsToWrite, pBlock->len,
pBlock->numOfCols, pBlock->minKey.ts, pBlock->maxKey.ts); pBlock->numOfCols, pBlock->minKey.ts, pBlock->maxKey.ts);
return 0; return 0;
...@@ -1313,7 +1305,7 @@ static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) { ...@@ -1313,7 +1305,7 @@ static int tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx) {
keyLimit = pBlock[1].minKey.ts - 1; keyLimit = pBlock[1].minKey.ts - 1;
} }
SSkipListIterator titer = *(pIter->pIter); STbDataIter titer = *(pIter->pIter);
if (tsdbLoadBlockDataCols(&(pCommith->readh), pBlock, NULL, &colId, 1, false) < 0) return -1; if (tsdbLoadBlockDataCols(&(pCommith->readh), pBlock, NULL, &colId, 1, false) < 0) return -1;
tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, &titer, keyLimit, INT32_MAX, NULL, tsdbLoadDataFromCache(TSDB_COMMIT_REPO(pCommith), pIter->pTable, &titer, keyLimit, INT32_MAX, NULL,
...@@ -1522,7 +1514,7 @@ static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *i ...@@ -1522,7 +1514,7 @@ static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *i
lastKey = key2; lastKey = key2;
} }
tSkipListIterNext(pCommitIter->pIter); tsdbTbDataIterNext(pCommitIter->pIter);
} else { } else {
if (lastKey != key1) { if (lastKey != key1) {
if (lastKey != TSKEY_INITIAL_VAL) { if (lastKey != TSKEY_INITIAL_VAL) {
...@@ -1554,7 +1546,7 @@ static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *i ...@@ -1554,7 +1546,7 @@ static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *i
tdAppendSTSRowToDataCol(row, pSchema, pTarget, true); tdAppendSTSRowToDataCol(row, pSchema, pTarget, true);
} }
++(*iter); ++(*iter);
tSkipListIterNext(pCommitIter->pIter); tsdbTbDataIterNext(pCommitIter->pIter);
} }
if (pTarget->numOfRows >= (maxRows - 1)) break; if (pTarget->numOfRows >= (maxRows - 1)) break;
......
source/dnode/vnode/src/tsdb/tsdbCommit2.c
浏览文件 @ 84aac9e3
...@@ -16,19 +16,19 @@ ...@@ -16,19 +16,19 @@
#include "tsdb.h" #include "tsdb.h"
typedef struct { typedef struct {
SMemTable *pMemTable; SMemTable2 *pMemTable;
int32_t minutes; int32_t minutes;
int8_t precision; int8_t precision;
TSKEY nCommitKey; TSKEY nCommitKey;
int32_t fid; int32_t fid;
TSKEY minKey; TSKEY minKey;
TSKEY maxKey; TSKEY maxKey;
SReadH readh; SReadH readh;
SDFileSet wSet; SDFileSet wSet;
SArray *aBlkIdx; SArray *aBlkIdx;
SArray *aSupBlk; SArray *aSupBlk;
SArray *aSubBlk; SArray *aSubBlk;
SArray *aDelInfo; SArray *aDelInfo;
} SCommitH; } SCommitH;
static int32_t tsdbCommitStart(SCommitH *pCHandle, STsdb *pTsdb); static int32_t tsdbCommitStart(SCommitH *pCHandle, STsdb *pTsdb);
...@@ -39,7 +39,7 @@ int32_t tsdbBegin2(STsdb *pTsdb) { ...@@ -39,7 +39,7 @@ int32_t tsdbBegin2(STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
ASSERT(pTsdb->mem == NULL); ASSERT(pTsdb->mem == NULL);
code = tsdbMemTableCreate2(pTsdb, (SMemTable **)&pTsdb->mem); code = tsdbMemTableCreate2(pTsdb, (SMemTable2 **)&pTsdb->mem);
if (code) { if (code) {
tsdbError("vgId:%d failed to begin TSDB since %s", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("vgId:%d failed to begin TSDB since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
goto _exit; goto _exit;
...@@ -80,8 +80,8 @@ _err: ...@@ -80,8 +80,8 @@ _err:
} }
static int32_t tsdbCommitStart(SCommitH *pCHandle, STsdb *pTsdb) { static int32_t tsdbCommitStart(SCommitH *pCHandle, STsdb *pTsdb) {
int32_t code = 0; int32_t code = 0;
SMemTable *pMemTable = (SMemTable *)pTsdb->mem; SMemTable2 *pMemTable = (SMemTable2 *)pTsdb->mem;
tsdbInfo("vgId:%d start to commit", TD_VID(pTsdb->pVnode)); tsdbInfo("vgId:%d start to commit", TD_VID(pTsdb->pVnode));
...@@ -131,9 +131,9 @@ _err: ...@@ -131,9 +131,9 @@ _err:
} }
static int32_t tsdbCommitEnd(SCommitH *pCHandle) { static int32_t tsdbCommitEnd(SCommitH *pCHandle) {
int32_t code = 0; int32_t code = 0;
STsdb *pTsdb = pCHandle->pMemTable->pTsdb; STsdb *pTsdb = pCHandle->pMemTable->pTsdb;
SMemTable *pMemTable = (SMemTable *)pTsdb->imem; SMemTable2 *pMemTable = (SMemTable2 *)pTsdb->imem;
// end transaction // end transaction
code = tsdbEndFSTxn(pTsdb); code = tsdbEndFSTxn(pTsdb);
......
source/dnode/vnode/src/tsdb/tsdbFile.c
浏览文件 @ 84aac9e3
...@@ -25,8 +25,7 @@ static const char *TSDB_FNAME_SUFFIX[] = { ...@@ -25,8 +25,7 @@ static const char *TSDB_FNAME_SUFFIX[] = {
"meta", // TSDB_FILE_META "meta", // TSDB_FILE_META
}; };
static void tsdbGetFilename(int vid, int fid, uint32_t ver, TSDB_FILE_T ftype, const char *dname, char *fname); static void tsdbGetFilename(int vid, int fid, uint32_t ver, TSDB_FILE_T ftype, const char *dname, char *fname);
// static int tsdbRollBackMFile(SMFile *pMFile);
static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo); static int tsdbEncodeDFInfo(void **buf, SDFInfo *pInfo);
static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo); static void *tsdbDecodeDFInfo(void *buf, SDFInfo *pInfo);
static int tsdbRollBackDFile(SDFile *pDFile); static int tsdbRollBackDFile(SDFile *pDFile);
......
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @ 84aac9e3
...@@ -15,67 +15,299 @@ ...@@ -15,67 +15,299 @@
#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, STsdbMemTable **ppMemTable) { #define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
STsdbMemTable *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 = (STsdbMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable));
if (pMemTable == NULL) { if (pMemTable == NULL) {
return -1; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
} }
pMemTable->pPool = pTsdb->pVnode->inUse;
T_REF_INIT_VAL(pMemTable, 1);
taosInitRWLatch(&pMemTable->latch); taosInitRWLatch(&pMemTable->latch);
pMemTable->keyMin = TSKEY_MAX; pMemTable->pTsdb = pTsdb;
pMemTable->keyMax = TSKEY_MIN; pMemTable->nRef = 1;
pMemTable->minKey = (TSDBKEY){.ts = TSKEY_MAX, .version = INT64_MAX};
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); *ppMemTable = pMemTable;
if (pMemTable->pHashIdx == NULL) { return code;
tSkipListDestroy(pMemTable->pSlIdx);
_err:
*ppMemTable = NULL;
return code;
}
void tsdbMemTableDestroy(SMemTable *pMemTable) {
if (pMemTable) {
taosArrayDestroy(pMemTable->aTbData);
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; 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;
} }
*ppMemTable = pMemTable;
return 0; 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;
}
void tsdbMemTableDestroy(STsdbMemTable *pMemTable) { int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock,
if (pMemTable) { SSubmitBlkRsp *pRsp) {
taosHashCleanup(pMemTable->pHashIdx); int32_t code = 0;
SSkipListIterator *pIter = tSkipListCreateIter(pMemTable->pSlIdx); SMemTable *pMemTable = pTsdb->mem;
SSkipListNode *pNode = NULL; STbData *pTbData = NULL;
STbData *pTbData = NULL; tb_uid_t suid = pMsgIter->suid;
for (;;) { tb_uid_t uid = pMsgIter->uid;
if (!tSkipListIterNext(pIter)) break; int32_t sverNew;
pNode = tSkipListIterGet(pIter);
pTbData = (STbData *)pNode->pData; // check if table exists (todo: refact)
tsdbFreeTbData(pTbData); 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;
} }
tSkipListDestroyIter(pIter);
tSkipListDestroy(pMemTable->pSlIdx);
taosMemoryFree(pMemTable);
} }
tGetTSDBRow((uint8_t *)SL_NODE_DATA(pIter->pNode), pRow);
return true;
} }
/** /**
...@@ -89,7 +321,7 @@ void tsdbMemTableDestroy(STsdbMemTable *pMemTable) { ...@@ -89,7 +321,7 @@ void tsdbMemTableDestroy(STsdbMemTable *pMemTable) {
* *
* The function tries to procceed AS MUCH AS POSSIBLE. * The function tries to procceed AS MUCH AS POSSIBLE.
*/ */
int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, STbDataIter *pIter, TSKEY maxKey, int maxRowsToRead,
SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo) { SDataCols *pCols, TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo) {
ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0); ASSERT(maxRowsToRead > 0 && nFilterKeys >= 0);
if (pIter == NULL) return 0; if (pIter == NULL) return 0;
...@@ -172,7 +404,7 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter ...@@ -172,7 +404,7 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter
} }
} }
tSkipListIterNext(pIter); tsdbTbDataIterNext(pIter);
row = tsdbNextIterRow(pIter); row = tsdbNextIterRow(pIter);
if (row == NULL || TD_ROW_KEY(row) > maxKey) { if (row == NULL || TD_ROW_KEY(row) > maxKey) {
rowKey = INT64_MAX; rowKey = INT64_MAX;
...@@ -212,7 +444,7 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter ...@@ -212,7 +444,7 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter
} }
} }
tSkipListIterNext(pIter); tsdbTbDataIterNext(pIter);
row = tsdbNextIterRow(pIter); row = tsdbNextIterRow(pIter);
if (row == NULL || TD_ROW_KEY(row) > maxKey) { if (row == NULL || TD_ROW_KEY(row) > maxKey) {
rowKey = INT64_MAX; rowKey = INT64_MAX;
...@@ -238,151 +470,281 @@ int tsdbLoadDataFromCache(STsdb *pTsdb, STable *pTable, SSkipListIterator *pIter ...@@ -238,151 +470,281 @@ 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;
STsdbMemTable *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; // get
TSKEY keyMax; idx = taosArraySearchIdx(pMemTable->aTbData, &pTbDataT, tbDataPCmprFn, TD_GE);
SSubmitBlk *pBlkCopy; if (idx >= 0) {
int64_t sverNew; pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, idx);
if (tbDataPCmprFn(&pTbDataT, &pTbData) == 0) goto _exit;
// check if table exists
SMetaReader mr = {0};
SMetaEntry me = {0};
metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0);
if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) {
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;
int8_t maxLevel = pMemTable->pTsdb->pVnode->config.tsdbCfg.slLevel;
pTbData = vnodeBufPoolMalloc(pPool, sizeof(*pTbData) + SL_NODE_SIZE(maxLevel) * 2);
if (pTbData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
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;
SL_NODE_BACKWARD(pTbData->sl.pHead, iLevel) = NULL;
SL_NODE_FORWARD(pTbData->sl.pTail, iLevel) = NULL;
}
void *p;
if (idx < 0) {
p = taosArrayPush(pMemTable->aTbData, &pTbData);
} else { } else {
metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid); p = taosArrayInsert(pMemTable->aTbData, idx, &pTbData);
sverNew = mr.me.stbEntry.schemaRow.version;
} }
metaReaderClear(&mr); if (p == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
_exit:
*ppTbData = pTbData;
return code;
_err:
*ppTbData = NULL;
return code;
}
// create container is nedd static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags) {
tptr = taosHashGet(pMemTable->pHashIdx, &(pMsgIter->uid), sizeof(pMsgIter->uid)); SMemSkipListNode *px;
if (tptr == NULL) { SMemSkipListNode *pn;
pTbData = tsdbNewTbData(pMsgIter->uid); TSDBKEY *pTKey;
if (pTbData == NULL) { int c;
return -1; 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];
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_BACKWARD(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_BACKWARD(px, iLevel);
}
}
// Put into skiplist pos[iLevel] = px;
tSkipListPut(pMemTable->pSlIdx, pTbData); }
}
} else { } else {
pTbData = *(STbData **)tptr; px = pTbData->sl.pHead;
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.pTail) {
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;
}
}
} }
}
// copy data to buffer pool static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
int32_t tlen = pMsgIter->dataLen + pMsgIter->schemaLen + sizeof(*pBlock); int8_t level = 1;
pBlkCopy = (SSubmitBlk *)vnodeBufPoolMalloc(pTsdb->mem->pPool, tlen); int8_t tlevel = TMIN(pSl->maxLevel, pSl->level + 1);
memcpy(pBlkCopy, pBlock, tlen); const uint32_t factor = 4;
tInitSubmitBlkIter(pMsgIter, pBlkCopy, &blkIter); while ((taosRandR(&pSl->seed) % factor) == 0 && level < tlevel) {
if (blkIter.row == NULL) return 0; level++;
keyMin = TD_ROW_KEY(blkIter.row); }
tSkipListPutBatchByIter(pTbData->pData, &blkIter, (iter_next_fn_t)tGetSubmitBlkNext); 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;
}
#ifdef TD_DEBUG_PRINT_ROW tPutTSDBRow((uint8_t *)SL_NODE_DATA(pNode), pRow);
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 // put
keyMax = TD_ROW_KEY(blkIter.row); for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
SMemSkipListNode *px = pos[iLevel];
pTbData->nrows += pMsgIter->numOfRows; if (forward) {
if (pTbData->keyMin > keyMin) pTbData->keyMin = keyMin; SMemSkipListNode *pNext = SL_NODE_FORWARD(px, iLevel);
if (pTbData->keyMax < keyMax) pTbData->keyMax = keyMax;
pMemTable->nRow += pMsgIter->numOfRows; SL_NODE_FORWARD(pNode, iLevel) = pNext;
if (pMemTable->keyMin > keyMin) pMemTable->keyMin = keyMin; SL_NODE_BACKWARD(pNode, iLevel) = px;
if (pMemTable->keyMax < keyMax) pMemTable->keyMax = keyMax;
pRsp->numOfRows = pMsgIter->numOfRows; SL_NODE_BACKWARD(pNext, iLevel) = pNode;
pRsp->affectedRows = pMsgIter->numOfRows; SL_NODE_FORWARD(px, iLevel) = pNode;
pRsp->sver = sverNew; } else {
SMemSkipListNode *pPrev = SL_NODE_BACKWARD(px, iLevel);
return 0; SL_NODE_FORWARD(pNode, iLevel) = px;
} SL_NODE_BACKWARD(pNode, iLevel) = pPrev;
static STbData *tsdbNewTbData(tb_uid_t uid) { SL_NODE_FORWARD(pPrev, iLevel) = pNode;
STbData *pTbData = (STbData *)taosMemoryCalloc(1, sizeof(*pTbData)); SL_NODE_BACKWARD(px, iLevel) = pNode;
if (pTbData == NULL) { }
return NULL;
} }
pTbData->uid = uid; pTbData->sl.size++;
pTbData->keyMin = TSKEY_MAX; if (pTbData->sl.level < pNode->level) {
pTbData->keyMax = TSKEY_MIN; pTbData->sl.level = pNode->level;
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; _exit:
return code;
} }
static void tsdbFreeTbData(STbData *pTbData) { static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version,
if (pTbData) { SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) {
tSkipListDestroy(pTbData->pData); int32_t code = 0;
taosMemoryFree(pTbData); 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;
} }
}
static char *tsdbGetTsTupleKey(const void *data) { return (char *)TD_ROW_KEY_ADDR((STSRow *)data); } if (tsdbKeyCmprFn(&key, &pTbData->minKey) < 0) {
pTbData->minKey = key;
}
static int tsdbTbDataComp(const void *arg1, const void *arg2) { if (tsdbKeyCmprFn(&key, &pMemTable->minKey) < 0) {
STbData *pTbData1 = (STbData *)arg1; pMemTable->minKey = key;
STbData *pTbData2 = (STbData *)arg2; }
if (pTbData1->uid > pTbData2->uid) { // forward put rest data
return 1; row.pTSRow = tGetSubmitBlkNext(&blkIter);
} else if (pTbData1->uid == pTbData2->uid) { if (row.pTSRow) {
return 0; for (int8_t iLevel = 0; iLevel < pTbData->sl.maxLevel; iLevel++) {
} else { pos[iLevel] = SL_NODE_BACKWARD(pos[iLevel], iLevel);
return -1; }
do {
key.ts = row.pTSRow->ts;
nRow++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
code = tbDataDoPut(pMemTable, pTbData, pos, &row, 1);
if (code) {
goto _err;
}
row.pTSRow = tGetSubmitBlkNext(&blkIter);
} 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 char *tsdbTbDataGetUid(const void *arg) { static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow) {
STbData *pTbData = (STbData *)arg; int32_t n = 0;
return (char *)(&(pTbData->uid));
n += tPutI64(p, pRow->version);
if (p) memcpy(p + n, pRow->pTSRow, pRow->pTSRow->len);
n += pRow->pTSRow->len;
return n;
} }
static int tsdbAppendTableRowToCols(STsdb *pTsdb, STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow *row, static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow) {
bool merge) { int32_t n = 0;
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); n += tGetI64(p, &pRow->version);
} pRow->pTSRow = (STSRow *)(p + n);
n += pRow->pTSRow->len;
return 0; return n;
} }
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
浏览文件 @ 84aac9e3
...@@ -15,11 +15,6 @@ ...@@ -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;
...@@ -35,21 +30,21 @@ typedef struct { ...@@ -35,21 +30,21 @@ typedef struct {
#define SL_MOVE_BACKWARD 0x1 #define SL_MOVE_BACKWARD 0x1
#define SL_MOVE_FROM_POS 0x2 #define SL_MOVE_FROM_POS 0x2
static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData); static int32_t tsdbGetOrCreateMemData(SMemTable2 *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData);
static int memDataPCmprFn(const void *p1, const void *p2); static int memDataPCmprFn(const void *p1, const void *p2);
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow); static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow); static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow);
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl); static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, SMemData *pMemData, int64_t version, static int32_t tsdbInsertTableDataImpl(SMemTable2 *pMemTable, SMemData *pMemData, int64_t version,
SVSubmitBlk *pSubmitBlk); SVSubmitBlk *pSubmitBlk);
static void memDataMovePosTo(SMemData *pMemData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags); static void memDataMovePosTo(SMemData *pMemData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags);
// SMemTable ============================================== // SMemTable ==============================================
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable) { int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable2 **ppMemTable) {
int32_t code = 0; int32_t code = 0;
SMemTable *pMemTable = NULL; SMemTable2 *pMemTable = NULL;
pMemTable = (SMemTable *)taosMemoryCalloc(1, sizeof(*pMemTable)); pMemTable = (SMemTable2 *)taosMemoryCalloc(1, sizeof(*pMemTable));
if (pMemTable == NULL) { if (pMemTable == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
...@@ -75,16 +70,16 @@ _err: ...@@ -75,16 +70,16 @@ _err:
return code; return code;
} }
void tsdbMemTableDestroy2(SMemTable *pMemTable) { void tsdbMemTableDestroy2(SMemTable2 *pMemTable) {
taosArrayDestroyEx(pMemTable->aMemData, NULL /*TODO*/); taosArrayDestroyEx(pMemTable->aMemData, NULL /*TODO*/);
taosMemoryFree(pMemTable); taosMemoryFree(pMemTable);
} }
int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmitBlk) { int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmitBlk) {
int32_t code = 0; int32_t code = 0;
SMemTable *pMemTable = (SMemTable *)pTsdb->mem; // TODO SMemTable2 *pMemTable = (SMemTable2 *)pTsdb->mem; // TODO
SMemData *pMemData; SMemData *pMemData;
TSDBROW row = {.version = version}; TSDBROW row = {.version = version};
ASSERT(pMemTable); ASSERT(pMemTable);
ASSERT(pSubmitBlk->nData > 0); ASSERT(pSubmitBlk->nData > 0);
...@@ -112,10 +107,10 @@ _err: ...@@ -112,10 +107,10 @@ _err:
} }
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) {
int32_t code = 0; int32_t code = 0;
SMemTable *pMemTable = (SMemTable *)pTsdb->mem; // TODO SMemTable2 *pMemTable = (SMemTable2 *)pTsdb->mem; // TODO
SMemData *pMemData; SMemData *pMemData;
SVBufPool *pPool = pTsdb->pVnode->inUse; SVBufPool *pPool = pTsdb->pVnode->inUse;
ASSERT(pMemTable); ASSERT(pMemTable);
...@@ -250,7 +245,7 @@ void tsdbMemDataIterGet(SMemDataIter *pIter, TSDBROW **ppRow) { ...@@ -250,7 +245,7 @@ void tsdbMemDataIterGet(SMemDataIter *pIter, TSDBROW **ppRow) {
} }
} }
static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData) { static int32_t tsdbGetOrCreateMemData(SMemTable2 *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData) {
int32_t code = 0; int32_t code = 0;
int32_t idx = 0; int32_t idx = 0;
SMemData *pMemDataT = &(SMemData){.suid = suid, .uid = uid}; SMemData *pMemDataT = &(SMemData){.suid = suid, .uid = uid};
...@@ -421,7 +416,7 @@ static void memDataMovePosTo(SMemData *pMemData, SMemSkipListNode **pos, TSDBKEY ...@@ -421,7 +416,7 @@ static void memDataMovePosTo(SMemData *pMemData, SMemSkipListNode **pos, TSDBKEY
} }
} }
static int32_t memDataDoPut(SMemTable *pMemTable, SMemData *pMemData, SMemSkipListNode **pos, TSDBROW *pRow, static int32_t memDataDoPut(SMemTable2 *pMemTable, SMemData *pMemData, SMemSkipListNode **pos, TSDBROW *pRow,
int8_t forward) { int8_t forward) {
int32_t code = 0; int32_t code = 0;
int8_t level; int8_t level;
...@@ -475,7 +470,7 @@ _exit: ...@@ -475,7 +470,7 @@ _exit:
return code; return code;
} }
static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, SMemData *pMemData, int64_t version, static int32_t tsdbInsertTableDataImpl(SMemTable2 *pMemTable, SMemData *pMemData, int64_t version,
SVSubmitBlk *pSubmitBlk) { SVSubmitBlk *pSubmitBlk) {
int32_t code = 0; int32_t code = 0;
int32_t n = 0; int32_t n = 0;
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 84aac9e3
...@@ -67,15 +67,16 @@ enum { ...@@ -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 {
...@@ -107,6 +108,7 @@ typedef struct SBlockLoadSuppInfo { ...@@ -107,6 +108,7 @@ typedef struct SBlockLoadSuppInfo {
typedef struct STsdbReadHandle { typedef struct STsdbReadHandle {
STsdb* pTsdb; STsdb* pTsdb;
uint64_t suid;
SQueryFilePos cur; // current position SQueryFilePos cur; // current position
int16_t order; int16_t order;
STimeWindow window; // the primary query time window that applies to all queries STimeWindow window; // the primary query time window that applies to all queries
...@@ -200,8 +202,8 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load ...@@ -200,8 +202,8 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load
int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) { int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT* pHandle) {
STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle; STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
int64_t rows = 0; int64_t rows = 0;
STsdbMemTable* pMemTable = NULL; // pTsdbReadHandle->pMemTable; SMemTable* pMemTable = NULL; // pTsdbReadHandle->pMemTable;
if (pMemTable == NULL) { if (pMemTable == NULL) {
return rows; return rows;
} }
...@@ -237,6 +239,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S ...@@ -237,6 +239,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
STableKeyInfo* pKeyInfo = (STableKeyInfo*)taosArrayGet(pTableList->pTableList, j); STableKeyInfo* pKeyInfo = (STableKeyInfo*)taosArrayGet(pTableList->pTableList, j);
STableCheckInfo info = {.lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid}; STableCheckInfo info = {.lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid};
info.suid = pTsdbReadHandle->suid;
if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) { if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
info.lastKey = pTsdbReadHandle->window.skey; info.lastKey = pTsdbReadHandle->window.skey;
...@@ -265,8 +268,8 @@ static void resetCheckInfo(STsdbReadHandle* pTsdbReadHandle) { ...@@ -265,8 +268,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)) {
...@@ -387,6 +390,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* ...@@ -387,6 +390,7 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
pReadHandle->locateStart = false; pReadHandle->locateStart = false;
pReadHandle->loadType = pCond->type; pReadHandle->loadType = pCond->type;
pReadHandle->suid = pCond->suid;
pReadHandle->outputCapacity = 4096; //((STsdb*)tsdb)->config.maxRowsPerFileBlock; pReadHandle->outputCapacity = 4096; //((STsdb*)tsdb)->config.maxRowsPerFileBlock;
pReadHandle->loadExternalRow = pCond->loadExternalRows; pReadHandle->loadExternalRow = pCond->loadExternalRows;
pReadHandle->currentLoadExternalRows = pCond->loadExternalRows; pReadHandle->currentLoadExternalRows = pCond->loadExternalRows;
...@@ -658,7 +662,7 @@ tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableL ...@@ -658,7 +662,7 @@ tsdbReaderT tsdbQueryLastRow(SVnode* pVnode, SQueryTableDataCond* pCond, STableL
} }
#if 0 #if 0
tsdbReaderT tsdbQueryCacheLastT(STsdb *tsdb, SQueryTableDataCond *pCond, STableGroupInfo *groupList, uint64_t qId, STsdbMemTable* pMemRef) { tsdbReaderT tsdbQueryCacheLastT(STsdb *tsdb, SQueryTableDataCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemTable* pMemRef) {
STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) tsdbQueryTablesT(tsdb, pCond, groupList, qId, pMemRef); STsdbReadHandle *pTsdbReadHandle = (STsdbReadHandle*) tsdbQueryTablesT(tsdb, pCond, groupList, qId, pMemRef);
if (pTsdbReadHandle == NULL) { if (pTsdbReadHandle == NULL) {
return NULL; return NULL;
...@@ -752,23 +756,22 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe ...@@ -752,23 +756,22 @@ 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;
int8_t backward = (pHandle->order == TSDB_ORDER_DESC) ? 1 : 0;
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}, backward, &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}, backward, &pCheckInfo->iiter);
tSkipListCreateIterFromVal((*pIMem)->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
} }
} }
...@@ -777,22 +780,23 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe ...@@ -777,22 +780,23 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe
return false; return false;
} }
bool memEmpty = (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tSkipListIterNext(pCheckInfo->iter)); bool memEmpty =
bool imemEmpty = (pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tSkipListIterNext(pCheckInfo->iiter)); (pCheckInfo->iter == NULL) || (pCheckInfo->iter != NULL && !tsdbTbDataIterGet(pCheckInfo->iter, NULL));
bool imemEmpty =
(pCheckInfo->iiter == NULL) || (pCheckInfo->iiter != NULL && !tsdbTbDataIterGet(pCheckInfo->iiter, NULL));
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)->keyMin, (*pMem)->keyMax, 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 +809,14 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe ...@@ -805,15 +809,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)->keyMin, (*pIMem)->keyMax, 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 +831,23 @@ static bool initTableMemIterator(STsdbReadHandle* pHandle, STableCheckInfo* pChe ...@@ -828,31 +831,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 +884,7 @@ static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, ...@@ -889,7 +884,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 +898,17 @@ static TSKEY extractFirstTraverseKey(STableCheckInfo* pCheckInfo, int32_t order, ...@@ -903,28 +898,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 +950,7 @@ static STSRow* getSRowInTableMem(STableCheckInfo* pCheckInfo, int32_t order, int ...@@ -966,7 +950,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 +979,7 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) { ...@@ -995,7 +979,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 +987,11 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) { ...@@ -1003,11 +987,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 +999,14 @@ static bool moveToNextRowInMem(STableCheckInfo* pCheckInfo) { ...@@ -1015,14 +999,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;
} }
} }
...@@ -1126,7 +1110,7 @@ static int32_t loadBlockInfo(STsdbReadHandle* pTsdbReadHandle, int32_t index, in ...@@ -1126,7 +1110,7 @@ static int32_t loadBlockInfo(STsdbReadHandle* pTsdbReadHandle, int32_t index, in
STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, index); STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, index);
pCheckInfo->numOfBlocks = 0; pCheckInfo->numOfBlocks = 0;
STable table = {.uid = pCheckInfo->tableId, .tid = pCheckInfo->tableId}; STable table = {.uid = pCheckInfo->tableId, .suid = pCheckInfo->suid};
table.pSchema = pTsdbReadHandle->pSchema; table.pSchema = pTsdbReadHandle->pSchema;
if (tsdbSetReadTable(&pTsdbReadHandle->rhelper, &table) != TSDB_CODE_SUCCESS) { if (tsdbSetReadTable(&pTsdbReadHandle->rhelper, &table) != TSDB_CODE_SUCCESS) {
...@@ -2924,7 +2908,7 @@ static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) { ...@@ -2924,7 +2908,7 @@ static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
// current result is empty // current result is empty
if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey && if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey &&
pTsdbReadHandle->cur.rows == 0) { pTsdbReadHandle->cur.rows == 0) {
// STsdbMemTable* pMemRef = pTsdbReadHandle->pMemTable; // SMemTable* pMemRef = pTsdbReadHandle->pMemTable;
// doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef); // doGetExternalRow(pTsdbReadHandle, TSDB_PREV_ROW, pMemRef);
// doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef); // doGetExternalRow(pTsdbReadHandle, TSDB_NEXT_ROW, pMemRef);
...@@ -3222,7 +3206,7 @@ bool tsdbNextDataBlock(tsdbReaderT pHandle) { ...@@ -3222,7 +3206,7 @@ bool tsdbNextDataBlock(tsdbReaderT pHandle) {
} }
} }
// static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, STsdbMemTable* pMemRef) { // static int32_t doGetExternalRow(STsdbReadHandle* pTsdbReadHandle, int16_t type, SMemTable* pMemRef) {
// STsdbReadHandle* pSecQueryHandle = NULL; // STsdbReadHandle* pSecQueryHandle = NULL;
// //
// if (type == TSDB_PREV_ROW && pTsdbReadHandle->prev) { // if (type == TSDB_PREV_ROW && pTsdbReadHandle->prev) {
......
source/dnode/vnode/src/tsdb/tsdbReadImpl.c
浏览文件 @ 84aac9e3
...@@ -156,6 +156,24 @@ int tsdbLoadBlockIdx(SReadH *pReadh) { ...@@ -156,6 +156,24 @@ int tsdbLoadBlockIdx(SReadH *pReadh) {
return 0; return 0;
} }
static int32_t tsdbBlockIdxCmprFn(const void *p1, const void *p2) {
SBlockIdx *pBlockIdx1 = (SBlockIdx *)p1;
SBlockIdx *pBlockIdx2 = (SBlockIdx *)p2;
if (pBlockIdx1->suid < pBlockIdx2->suid) {
return -1;
} else if (pBlockIdx1->suid > pBlockIdx2->suid) {
return 1;
}
if (pBlockIdx1->uid < pBlockIdx2->uid) {
return -1;
} else if (pBlockIdx1->uid > pBlockIdx2->uid) {
return 1;
}
return 0;
}
int tsdbSetReadTable(SReadH *pReadh, STable *pTable) { int tsdbSetReadTable(SReadH *pReadh, STable *pTable) {
STSchema *pSchema = tsdbGetTableSchemaImpl(TSDB_READ_REPO(pReadh), pTable, false, false, -1); STSchema *pSchema = tsdbGetTableSchemaImpl(TSDB_READ_REPO(pReadh), pTable, false, false, -1);
...@@ -171,33 +189,40 @@ int tsdbSetReadTable(SReadH *pReadh, STable *pTable) { ...@@ -171,33 +189,40 @@ int tsdbSetReadTable(SReadH *pReadh, STable *pTable) {
return -1; return -1;
} }
size_t size = taosArrayGetSize(pReadh->aBlkIdx); uint8_t *p = taosArraySearch(pReadh->aBlkIdx, &(SBlockIdx){.suid = pTable->suid, .uid = pTable->uid},
if (size > 0) { tsdbBlockIdxCmprFn, TD_EQ);
while (true) { if (p == NULL) {
if (pReadh->cidx >= size) {
pReadh->pBlkIdx = NULL;
break;
}
SBlockIdx *pBlkIdx = taosArrayGet(pReadh->aBlkIdx, pReadh->cidx);
if (pBlkIdx->uid == TABLE_TID(pTable)) {
if (pBlkIdx->uid == TABLE_UID(pTable)) {
pReadh->pBlkIdx = pBlkIdx;
} else {
pReadh->pBlkIdx = NULL;
}
pReadh->cidx++;
break;
} else if (pBlkIdx->uid > TABLE_TID(pTable)) {
pReadh->pBlkIdx = NULL;
break;
} else {
pReadh->cidx++;
}
}
} else {
pReadh->pBlkIdx = NULL; pReadh->pBlkIdx = NULL;
} } else {
pReadh->pBlkIdx = (SBlockIdx *)p;
}
// size_t size = taosArrayGetSize(pReadh->aBlkIdx);
// if (size > 0) {
// while (true) {
// if (pReadh->cidx >= size) {
// pReadh->pBlkIdx = NULL;
// break;
// }
// SBlockIdx *pBlkIdx = taosArrayGet(pReadh->aBlkIdx, pReadh->cidx);
// if (pBlkIdx->uid == TABLE_TID(pTable)) {
// if (pBlkIdx->uid == TABLE_UID(pTable)) {
// pReadh->pBlkIdx = pBlkIdx;
// } else {
// pReadh->pBlkIdx = NULL;
// }
// pReadh->cidx++;
// break;
// } else if (pBlkIdx->uid > TABLE_TID(pTable)) {
// pReadh->pBlkIdx = NULL;
// break;
// } else {
// pReadh->cidx++;
// }
// }
// } else {
// pReadh->pBlkIdx = NULL;
// }
return 0; return 0;
} }
...@@ -553,12 +578,12 @@ static int tsdbLoadBlockOffset(SReadH *pReadh, SBlock *pBlock) { ...@@ -553,12 +578,12 @@ static int tsdbLoadBlockOffset(SReadH *pReadh, SBlock *pBlock) {
int tsdbEncodeSBlockIdx(void **buf, SBlockIdx *pIdx) { int tsdbEncodeSBlockIdx(void **buf, SBlockIdx *pIdx) {
int tlen = 0; int tlen = 0;
// tlen += taosEncodeVariantI32(buf, pIdx->tid); tlen += taosEncodeFixedU64(buf, pIdx->suid);
tlen += taosEncodeFixedU64(buf, pIdx->uid);
tlen += taosEncodeVariantU32(buf, pIdx->len); tlen += taosEncodeVariantU32(buf, pIdx->len);
tlen += taosEncodeVariantU32(buf, pIdx->offset); tlen += taosEncodeVariantU32(buf, pIdx->offset);
tlen += taosEncodeFixedU8(buf, pIdx->hasLast); tlen += taosEncodeFixedU8(buf, pIdx->hasLast);
tlen += taosEncodeVariantU32(buf, pIdx->numOfBlocks); tlen += taosEncodeVariantU32(buf, pIdx->numOfBlocks);
tlen += taosEncodeFixedU64(buf, pIdx->uid);
tlen += taosEncodeFixedU64(buf, pIdx->maxKey.ts); tlen += taosEncodeFixedU64(buf, pIdx->maxKey.ts);
return tlen; return tlen;
...@@ -570,6 +595,10 @@ void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx) { ...@@ -570,6 +595,10 @@ void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx) {
uint64_t value = 0; uint64_t value = 0;
// if ((buf = taosDecodeVariantI32(buf, &(pIdx->tid))) == NULL) return NULL; // if ((buf = taosDecodeVariantI32(buf, &(pIdx->tid))) == NULL) return NULL;
if ((buf = taosDecodeFixedU64(buf, &value)) == NULL) return NULL;
pIdx->suid = (int64_t)value;
if ((buf = taosDecodeFixedU64(buf, &value)) == NULL) return NULL;
pIdx->uid = (int64_t)value;
if ((buf = taosDecodeVariantU32(buf, &(pIdx->len))) == NULL) return NULL; if ((buf = taosDecodeVariantU32(buf, &(pIdx->len))) == NULL) return NULL;
if ((buf = taosDecodeVariantU32(buf, &(pIdx->offset))) == NULL) return NULL; if ((buf = taosDecodeVariantU32(buf, &(pIdx->offset))) == NULL) return NULL;
if ((buf = taosDecodeFixedU8(buf, &(hasLast))) == NULL) return NULL; if ((buf = taosDecodeFixedU8(buf, &(hasLast))) == NULL) return NULL;
...@@ -577,8 +606,6 @@ void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx) { ...@@ -577,8 +606,6 @@ void *tsdbDecodeSBlockIdx(void *buf, SBlockIdx *pIdx) {
if ((buf = taosDecodeVariantU32(buf, &(numOfBlocks))) == NULL) return NULL; if ((buf = taosDecodeVariantU32(buf, &(numOfBlocks))) == NULL) return NULL;
pIdx->numOfBlocks = numOfBlocks; pIdx->numOfBlocks = numOfBlocks;
if ((buf = taosDecodeFixedU64(buf, &value)) == NULL) return NULL; if ((buf = taosDecodeFixedU64(buf, &value)) == NULL) return NULL;
pIdx->uid = (int64_t)value;
if ((buf = taosDecodeFixedU64(buf, &value)) == NULL) return NULL;
pIdx->maxKey.ts = (TSKEY)value; pIdx->maxKey.ts = (TSKEY)value;
return buf; return buf;
......
source/dnode/vnode/src/tsdb/tsdbWrite.c
浏览文件 @ 84aac9e3
...@@ -39,7 +39,7 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp * ...@@ -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;
} }
......
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 84aac9e3
...@@ -799,7 +799,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq ...@@ -799,7 +799,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;
} }
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 84aac9e3
...@@ -4760,6 +4760,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi ...@@ -4760,6 +4760,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
pCond->numOfTWindows = 1; pCond->numOfTWindows = 1;
pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow)); pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow));
pCond->twindows[0] = pTableScanNode->scanRange; pCond->twindows[0] = pTableScanNode->scanRange;
pCond->suid = pTableScanNode->scan.suid;
#if 1 #if 1
// todo work around a problem, remove it later // todo work around a problem, remove it later
......
source/libs/parser/inc/parInsertData.h
浏览文件 @ 84aac9e3
...@@ -116,8 +116,7 @@ static FORCE_INLINE void getSTSRowAppendInfo(uint8_t rowType, SParsedDataColInfo ...@@ -116,8 +116,7 @@ static FORCE_INLINE void getSTSRowAppendInfo(uint8_t rowType, SParsedDataColInfo
} }
static FORCE_INLINE int32_t setBlockInfo(SSubmitBlk *pBlocks, STableDataBlocks *dataBuf, int32_t numOfRows) { static FORCE_INLINE int32_t setBlockInfo(SSubmitBlk *pBlocks, STableDataBlocks *dataBuf, int32_t numOfRows) {
pBlocks->suid = pBlocks->suid = (TSDB_NORMAL_TABLE == dataBuf->pTableMeta->tableType ? 0 : dataBuf->pTableMeta->suid);
(TSDB_NORMAL_TABLE == dataBuf->pTableMeta->tableType ? dataBuf->pTableMeta->uid : dataBuf->pTableMeta->suid);
pBlocks->uid = dataBuf->pTableMeta->uid; pBlocks->uid = dataBuf->pTableMeta->uid;
pBlocks->sversion = dataBuf->pTableMeta->sversion; pBlocks->sversion = dataBuf->pTableMeta->sversion;
pBlocks->schemaLen = dataBuf->createTbReqLen; pBlocks->schemaLen = dataBuf->createTbReqLen;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或