This commit is contained in:
Hongze Cheng 2020-09-16 11:22:49 +08:00
parent dee16dd45b
commit 09e91f9ad4
7 changed files with 602 additions and 540 deletions

View File

@ -411,9 +411,9 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
} }
if (cond.start != NULL) { if (cond.start != NULL) {
iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_ASC); iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_ASC);
} else { } else {
iter = tSkipListCreateIterFromVal(pSkipList, (char*)(cond.end ? cond.end->v: NULL), pSkipList->keyInfo.type, TSDB_ORDER_DESC); iter = tSkipListCreateIterFromVal(pSkipList, (char*)(cond.end ? cond.end->v: NULL), pSkipList->type, TSDB_ORDER_DESC);
} }
if (cond.start != NULL) { if (cond.start != NULL) {
@ -468,7 +468,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
tSkipListDestroyIter(iter); tSkipListDestroyIter(iter);
comp = true; comp = true;
iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_DESC); iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->type, TSDB_ORDER_DESC);
while(tSkipListIterNext(iter)) { while(tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter); SSkipListNode* pNode = tSkipListIterGet(iter);
comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0); comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);

View File

@ -37,38 +37,25 @@ static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables);
int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) { int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
STsdbCfg * pCfg = &pRepo->config; STsdbCfg * pCfg = &pRepo->config;
STsdbMeta * pMeta = pRepo->tsdbMeta; STsdbMeta * pMeta = pRepo->tsdbMeta;
int32_t level = 0;
int32_t headSize = 0;
TSKEY key = dataRowKey(row); TSKEY key = dataRowKey(row);
SMemTable * pMemTable = pRepo->mem; SMemTable * pMemTable = pRepo->mem;
STableData *pTableData = NULL; STableData *pTableData = NULL;
SSkipList * pSList = NULL; // SSkipList * pSList = NULL;
if (pMemTable != NULL && TABLE_TID(pTable) < pMemTable->maxTables && pMemTable->tData[TABLE_TID(pTable)] != NULL && // if (pMemTable != NULL && TABLE_TID(pTable) < pMemTable->maxTables && pMemTable->tData[TABLE_TID(pTable)] != NULL &&
pMemTable->tData[TABLE_TID(pTable)]->uid == TABLE_UID(pTable)) { // pMemTable->tData[TABLE_TID(pTable)]->uid == TABLE_UID(pTable)) {
pTableData = pMemTable->tData[TABLE_TID(pTable)]; // pTableData = pMemTable->tData[TABLE_TID(pTable)];
pSList = pTableData->pData; // pSList = pTableData->pData;
} // }
tSkipListNewNodeInfo(pSList, &level, &headSize);
SSkipListNode *pNode = (SSkipListNode *)malloc(headSize + sizeof(SDataRow *));
if (pNode == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
void *pRow = tsdbAllocBytes(pRepo, dataRowLen(row)); void *pRow = tsdbAllocBytes(pRepo, dataRowLen(row));
if (pRow == NULL) { if (pRow == NULL) {
tsdbError("vgId:%d failed to insert row with key %" PRId64 " to table %s while allocate %d bytes since %s", tsdbError("vgId:%d failed to insert row with key %" PRId64 " to table %s while allocate %d bytes since %s",
REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), dataRowLen(row), tstrerror(terrno)); REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), dataRowLen(row), tstrerror(terrno));
free(pNode);
return -1; return -1;
} }
pNode->level = level;
dataRowCpy(pRow, row); dataRowCpy(pRow, row);
*(SDataRow *)SL_GET_NODE_DATA(pNode) = pRow;
// Operations above may change pRepo->mem, retake those values // Operations above may change pRepo->mem, retake those values
ASSERT(pRepo->mem != NULL); ASSERT(pRepo->mem != NULL);
@ -77,7 +64,6 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
if (TABLE_TID(pTable) >= pMemTable->maxTables) { if (TABLE_TID(pTable) >= pMemTable->maxTables) {
if (tsdbAdjustMemMaxTables(pMemTable, pMeta->maxTables) < 0) { if (tsdbAdjustMemMaxTables(pMemTable, pMeta->maxTables) < 0) {
tsdbFreeBytes(pRepo, pRow, dataRowLen(row)); tsdbFreeBytes(pRepo, pRow, dataRowLen(row));
free(pNode);
return -1; return -1;
} }
} }
@ -97,7 +83,6 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
" to table %s while create new table data object since %s", " to table %s while create new table data object since %s",
REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), tstrerror(terrno)); REPO_ID(pRepo), key, TABLE_CHAR_NAME(pTable), tstrerror(terrno));
tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row)); tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row));
free(pNode);
return -1; return -1;
} }
@ -106,20 +91,19 @@ int tsdbInsertRowToMem(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
ASSERT((pTableData != NULL) && pTableData->uid == TABLE_UID(pTable)); ASSERT((pTableData != NULL) && pTableData->uid == TABLE_UID(pTable));
if (tSkipListPut(pTableData->pData, pNode) == NULL) { int64_t oldSize = SL_GET_SIZE(pTableData->pData);
if (tSkipListPut(pTableData->pData, (void *)(&pRow), sizeof(void *)) == NULL) {
tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row)); tsdbFreeBytes(pRepo, (void *)pRow, dataRowLen(row));
free(pNode);
} else { } else {
int64_t deltaSize = SL_GET_SIZE(pTableData->pData) - oldSize;
if (TABLE_LASTKEY(pTable) < key) TABLE_LASTKEY(pTable) = key; if (TABLE_LASTKEY(pTable) < key) TABLE_LASTKEY(pTable) = key;
if (pMemTable->keyFirst > key) pMemTable->keyFirst = key; if (pMemTable->keyFirst > key) pMemTable->keyFirst = key;
if (pMemTable->keyLast < key) pMemTable->keyLast = key; if (pMemTable->keyLast < key) pMemTable->keyLast = key;
pMemTable->numOfRows++; pMemTable->numOfRows += deltaSize;
if (pTableData->keyFirst > key) pTableData->keyFirst = key; if (pTableData->keyFirst > key) pTableData->keyFirst = key;
if (pTableData->keyLast < key) pTableData->keyLast = key; if (pTableData->keyLast < key) pTableData->keyLast = key;
pTableData->numOfRows++; pTableData->numOfRows += deltaSize;
ASSERT(pTableData->numOfRows == tSkipListGetSize(pTableData->pData));
} }
tsdbTrace("vgId:%d a row is inserted to table %s tid %d uid %" PRIu64 " key %" PRIu64, REPO_ID(pRepo), tsdbTrace("vgId:%d a row is inserted to table %s tid %d uid %" PRIu64 " key %" PRIu64, REPO_ID(pRepo),
@ -439,7 +423,7 @@ static STableData *tsdbNewTableData(STsdbCfg *pCfg, STable *pTable) {
pTableData->numOfRows = 0; pTableData->numOfRows = 0;
pTableData->pData = tSkipListCreate(TSDB_DATA_SKIPLIST_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, pTableData->pData = tSkipListCreate(TSDB_DATA_SKIPLIST_LEVEL, TSDB_DATA_TYPE_TIMESTAMP,
TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 1, tsdbGetTsTupleKey); TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], /*SL_DISCARD_DUP_KEY*/ SL_APPEND_DUP_KEY, tsdbGetTsTupleKey);
if (pTableData->pData == NULL) { if (pTableData->pData == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
goto _err; goto _err;

View File

@ -690,7 +690,7 @@ static STable *tsdbNewTable(STableCfg *pCfg, bool isSuper) {
} }
pTable->tagVal = NULL; pTable->tagVal = NULL;
STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN); STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN);
pTable->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), 1, 0, 1, getTagIndexKey); pTable->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), SL_ALLOW_DUP_KEY, getTagIndexKey);
if (pTable->pIndex == NULL) { if (pTable->pIndex == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
goto _err; goto _err;
@ -892,23 +892,8 @@ static int tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable, bool refSuper
pTable->pSuper = pSTable; pTable->pSuper = pSTable;
int32_t level = 0; tSkipListPut(pSTable->pIndex, (void *)(&pTable), sizeof(STable *));
int32_t headSize = 0;
tSkipListNewNodeInfo(pSTable->pIndex, &level, &headSize);
// NOTE: do not allocate the space for key, since in each skip list node, only keep the pointer to pTable, not the
// actual key value, and the key value will be retrieved during query through the pTable and getTagIndexKey function
SSkipListNode *pNode = calloc(1, headSize + sizeof(STable *));
if (pNode == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
pNode->level = level;
memcpy(SL_GET_NODE_DATA(pNode), &pTable, sizeof(STable *));
tSkipListPut(pSTable->pIndex, pNode);
if (refSuper) T_REF_INC(pSTable); if (refSuper) T_REF_INC(pSTable);
return 0; return 0;
} }
@ -1165,7 +1150,7 @@ static void *tsdbDecodeTable(void *buf, STable **pRTable) {
buf = tdDecodeSchema(buf, &(pTable->tagSchema)); buf = tdDecodeSchema(buf, &(pTable->tagSchema));
STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN); STColumn *pCol = schemaColAt(pTable->tagSchema, DEFAULT_TAG_INDEX_COLUMN);
pTable->pIndex = pTable->pIndex =
tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), 1, 0, 1, getTagIndexKey); tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, colType(pCol), (uint8_t)(colBytes(pCol)), SL_ALLOW_DUP_KEY, getTagIndexKey);
if (pTable->pIndex == NULL) { if (pTable->pIndex == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbFreeTable(pTable); tsdbFreeTable(pTable);
@ -1191,7 +1176,7 @@ static int tsdbGetTableEncodeSize(int8_t act, STable *pTable) {
tlen = sizeof(SListNode) + sizeof(SActObj) + sizeof(SActCont) + tsdbEncodeTable(NULL, pTable) + sizeof(TSCKSUM); tlen = sizeof(SListNode) + sizeof(SActObj) + sizeof(SActCont) + tsdbEncodeTable(NULL, pTable) + sizeof(TSCKSUM);
} else { } else {
if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) { if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
tlen = (int)((sizeof(SListNode) + sizeof(SActObj)) * (tSkipListGetSize(pTable->pIndex) + 1)); tlen = (int)((sizeof(SListNode) + sizeof(SActObj)) * (SL_GET_SIZE(pTable->pIndex) + 1));
} else { } else {
tlen = sizeof(SListNode) + sizeof(SActObj); tlen = sizeof(SListNode) + sizeof(SActObj);
} }

View File

@ -7,7 +7,7 @@ TARGET_LINK_LIBRARIES(tutil pthread osdetail lz4)
IF (TD_LINUX) IF (TD_LINUX)
TARGET_LINK_LIBRARIES(tutil m rt) TARGET_LINK_LIBRARIES(tutil m rt)
ADD_SUBDIRECTORY(tests) # ADD_SUBDIRECTORY(tests)
FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/) FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/)
IF (ICONV_INCLUDE_EXIST) IF (ICONV_INCLUDE_EXIST)

View File

@ -27,6 +27,14 @@ extern "C" {
#define MAX_SKIP_LIST_LEVEL 15 #define MAX_SKIP_LIST_LEVEL 15
#define SKIP_LIST_RECORD_PERFORMANCE 0 #define SKIP_LIST_RECORD_PERFORMANCE 0
// For key property setting
#define SL_ALLOW_DUP_KEY (uint8_t)0x0 // Allow duplicate key exists
#define SL_DISCARD_DUP_KEY (uint8_t)0x1 // Discard duplicate key
#define SL_UPDATA_DUP_KEY (uint8_t)0x2 // Update duplicate key by remove/insert
#define SL_APPEND_DUP_KEY (uint8_t)0x3 // Update duplicate key by append
// For thread safety setting
#define SL_THREAD_SAFE (uint8_t)0x4
typedef char *SSkipListKey; typedef char *SSkipListKey;
typedef char *(*__sl_key_fn_t)(const void *); typedef char *(*__sl_key_fn_t)(const void *);
@ -41,6 +49,9 @@ typedef struct SSkipListNode {
uint8_t level; uint8_t level;
} SSkipListNode; } SSkipListNode;
#define SL_IS_THREAD_SAFE(flags) ((flags)&SL_THREAD_SAFE)
#define SL_DUP_MODE(flags) ((flags) & ((((uint8_t)1) << 2) - 1))
#define SL_NODE_HEADER_SIZE(_l) (sizeof(SSkipListNode) + ((_l) << 1u) * POINTER_BYTES) #define SL_NODE_HEADER_SIZE(_l) (sizeof(SSkipListNode) + ((_l) << 1u) * POINTER_BYTES)
#define SL_GET_FORWARD_POINTER(n, _l) ((SSkipListNode **)((char *)(n) + sizeof(SSkipListNode)))[(_l)] #define SL_GET_FORWARD_POINTER(n, _l) ((SSkipListNode **)((char *)(n) + sizeof(SSkipListNode)))[(_l)]
@ -54,6 +65,8 @@ typedef struct SSkipListNode {
#define SL_GET_SL_MAX_KEY(s) (SL_GET_NODE_KEY((s), SL_GET_BACKWARD_POINTER((s)->pTail, 0))) #define SL_GET_SL_MAX_KEY(s) (SL_GET_NODE_KEY((s), SL_GET_BACKWARD_POINTER((s)->pTail, 0)))
#define SL_GET_NODE_LEVEL(n) *(uint8_t *)((n)) #define SL_GET_NODE_LEVEL(n) *(uint8_t *)((n))
#define SL_GET_SIZE(s) (s)->size
#define SL_GET_TSIZE(s) (s)->tsize
/* /*
* @version 0.3 * @version 0.3
@ -113,11 +126,14 @@ typedef struct SSkipListKeyInfo {
typedef struct SSkipList { typedef struct SSkipList {
__compar_fn_t comparFn; __compar_fn_t comparFn;
__sl_key_fn_t keyFn; __sl_key_fn_t keyFn;
uint32_t size;
uint8_t maxLevel;
uint8_t level;
SSkipListKeyInfo keyInfo;
pthread_rwlock_t *lock; pthread_rwlock_t *lock;
uint16_t len;
uint8_t maxLevel;
uint8_t flags;
uint8_t type; // static info above
uint8_t level;
uint32_t size; // not including duplicate keys
uint32_t tsize; // including duplicate keys
SSkipListNode * pHead; // point to the first element SSkipListNode * pHead; // point to the first element
SSkipListNode * pTail; // point to the last element SSkipListNode * pTail; // point to the last element
#if SKIP_LIST_RECORD_PERFORMANCE #if SKIP_LIST_RECORD_PERFORMANCE
@ -145,8 +161,7 @@ typedef struct SSkipListIterator {
* @param dupKey allow the duplicated key in the skip list * @param dupKey allow the duplicated key in the skip list
* @return * @return
*/ */
SSkipList *tSkipListCreate(uint8_t nMaxLevel, uint8_t keyType, uint8_t keyLen, uint8_t dupKey, uint8_t threadsafe, SSkipList *tSkipListCreate(uint8_t nMaxLevel, uint8_t keyType, uint16_t keyLen, uint8_t flags, __sl_key_fn_t fn);
uint8_t freeNode, __sl_key_fn_t fn);
/** /**
* *
@ -163,6 +178,17 @@ void *tSkipListDestroy(SSkipList *pSkipList);
*/ */
void tSkipListNewNodeInfo(SSkipList *pSkipList, int32_t *level, int32_t *headSize); void tSkipListNewNodeInfo(SSkipList *pSkipList, int32_t *level, int32_t *headSize);
/**
* put the data into the skiplist
* If failed, NULL will be returned, otherwise, the pNode will be returned.
*
* @param pSkipList
* @param pData
* @param dataLen
* @return
*/
SSkipListNode *tSkipListPut(SSkipList *pSkipList, void *pData, int dataLen);
/** /**
* put the skip list node into the skip list. * put the skip list node into the skip list.
* If failed, NULL will be returned, otherwise, the pNode will be returned. * If failed, NULL will be returned, otherwise, the pNode will be returned.
@ -171,7 +197,7 @@ void tSkipListNewNodeInfo(SSkipList *pSkipList, int32_t *level, int32_t *headSiz
* @param pNode * @param pNode
* @return * @return
*/ */
SSkipListNode *tSkipListPut(SSkipList *pSkipList, SSkipListNode *pNode); SSkipListNode *tSkipListPutNode(SSkipList *pSkipList, SSkipListNode *pNode);
/** /**
* get *all* nodes which key are equivalent to pKey * get *all* nodes which key are equivalent to pKey
@ -182,13 +208,6 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, SSkipListNode *pNode);
*/ */
SArray *tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey); SArray *tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey);
/**
* get the size of skip list
* @param pSkipList
* @return
*/
size_t tSkipListGetSize(const SSkipList *pSkipList);
/** /**
* display skip list of the given level, for debug purpose only * display skip list of the given level, for debug purpose only
* @param pSkipList * @param pSkipList

File diff suppressed because it is too large Load Diff

View File

@ -247,7 +247,7 @@ void skiplistPerformanceTest() {
printf("total:%" PRIu64 " ms, avg:%f\n", e - s, (e - s) / (double)size); printf("total:%" PRIu64 " ms, avg:%f\n", e - s, (e - s) / (double)size);
printf("max level of skiplist:%d, actually level:%d\n ", pSkipList->maxLevel, pSkipList->level); printf("max level of skiplist:%d, actually level:%d\n ", pSkipList->maxLevel, pSkipList->level);
assert(tSkipListGetSize(pSkipList) == size); assert(SL_GET_SIZE(pSkipList) == size);
// printf("the level of skiplist is:\n"); // printf("the level of skiplist is:\n");
// //
@ -273,7 +273,7 @@ void skiplistPerformanceTest() {
int64_t et = taosGetTimestampMs(); int64_t et = taosGetTimestampMs();
printf("delete %d data from skiplist, elapased time:%" PRIu64 "ms\n", 10000, et - st); printf("delete %d data from skiplist, elapased time:%" PRIu64 "ms\n", 10000, et - st);
assert(tSkipListGetSize(pSkipList) == size); assert(SL_GET_SIZE(pSkipList) == size);
tSkipListDestroy(pSkipList); tSkipListDestroy(pSkipList);
taosTFree(total); taosTFree(total);