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

Merge remote-tracking branch 'origin/3.0' into feature/dnode

This commit is contained in:
Shengliang Guan 2022-05-05 16:35:25 +08:00
parent d2f889a3a0 e882eea6a3
commit abcdeb8418
27 changed files with 1227 additions and 1964 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmake/cmake.define
View File

@ -67,7 +67,13 @@ ELSE ()
IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64" OR ${CMAKE_SYSTEM_PROCESSOR} MATCHES "aarch64")
ADD_DEFINITIONS("-D_TD_ARM_")
ELSE ()
ADD_DEFINITIONS("-msse4.2 -mfma")
ADD_DEFINITIONS("-msse4.2")
IF("${FMA_SUPPORT}" MATCHES "true")
MESSAGE(STATUS "turn fma function support on")
ADD_DEFINITIONS("-mfma")
ELSE ()
MESSAGE(STATUS "turn fma function support off")
ENDIF()
ENDIF ()
ENDIF ()

12
include/libs/index/index.h
View File

@ -47,7 +47,17 @@ typedef enum {
} SIndexOperOnColumn;
typedef enum { MUST = 0, SHOULD, NOT } EIndexOperatorType;
typedef enum { QUERY_TERM = 0, QUERY_PREFIX, QUERY_SUFFIX, QUERY_REGEX, QUERY_RANGE } EIndexQueryType;
typedef enum {
QUERY_TERM = 0,
QUERY_PREFIX,
QUERY_SUFFIX,
QUERY_REGEX,
QUERY_LESS_THAN,
QUERY_LESS_EQUAL,
QUERY_GREATER_THAN,
QUERY_GREATER_EQUAL,
QUERY_RANGE
} EIndexQueryType;
/*
* create multi query

70
include/util/tskiplist2.h Normal file
View File

@ -0,0 +1,70 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_UTIL_SKIPLIST2_H_
#define _TD_UTIL_SKIPLIST2_H_
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
#define SL_MAX_LEVEL 15
typedef struct SSkipList2 SSkipList2;
typedef struct SSLCursor SSLCursor;
typedef struct SSLCfg SSLCfg;
typedef struct SSLNode SSLNode;
typedef int32_t (*tslCmprFn)(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2);
// SSkipList2
int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl);
int32_t slClose(SSkipList2 *pSl);
int32_t slClear(SSkipList2 *pSl);
// SSLCursor
int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc);
int32_t slcClose(SSLCursor *pSlc);
int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey);
int32_t slcMoveToNext(SSLCursor *pSlc);
int32_t slcMoveToPrev(SSLCursor *pSlc);
int32_t slcMoveToFirst(SSLCursor *pSlc);
int32_t slcMoveToLast(SSLCursor *pSlc);
int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData);
int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData);
int32_t slcDrop(SSLCursor *pSlc);
// struct
struct SSLCfg {
int8_t maxLevel;
int32_t nKey;
int32_t nData;
tslCmprFn cmprFn;
void *pPool;
void *(*xMalloc)(void *, int32_t size);
void (*xFree)(void *, void *);
};
struct SSLCursor {
SSkipList2 *pSl;
SSLNode **forwards[SL_MAX_LEVEL];
};
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_SKIPLIST2_H_*/

18
source/client/src/clientStmt.c
View File

@ -649,6 +649,19 @@ int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
}
if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
}
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
@ -658,8 +671,11 @@ int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
pStmt->exec.pRequest->body.pDag = NULL;
STMT_ERR_RET(stmtBackupQueryFields(pStmt));
} else {
STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
}
*nums = taosArrayGetSize(pStmt->sql.pQueryPlan->pPlaceholderValues);
} else {
STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));

2
source/dnode/vnode/src/inc/meta.h
View File

@ -103,8 +103,6 @@ typedef struct {
#if 1
// int metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
int metaDropTable(SMeta* pMeta, tb_uid_t uid);
SMSmaCursor* metaOpenSmaCursor(SMeta* pMeta, tb_uid_t uid);
void metaCloseSmaCursor(SMSmaCursor* pSmaCur);
int64_t metaSmaCursorNext(SMSmaCursor* pSmaCur);

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

@ -46,13 +46,13 @@
extern "C" {
#endif
typedef struct SVnodeInfo SVnodeInfo;
typedef struct SMeta SMeta;
typedef struct STsdb STsdb;
typedef struct STQ STQ;
typedef struct SVState SVState;
typedef struct SVBufPool SVBufPool;
typedef struct SQWorker SQHandle;
typedef struct SVnodeInfo SVnodeInfo;
typedef struct SMeta SMeta;
typedef struct STsdb STsdb;
typedef struct STQ STQ;
typedef struct SVState SVState;
typedef struct SVBufPool SVBufPool;
typedef struct SQWorker SQHandle;
#define VNODE_META_DIR "meta"
#define VNODE_TSDB_DIR "tsdb"
@ -77,6 +77,7 @@ int metaCommit(SMeta* pMeta);
int metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
int metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq);
int metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq);
int metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq);
SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
STSchema* metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
int metaGetTableEntryByName(SMetaReader* pReader, const char* name);
@ -100,7 +101,7 @@ int32_t tsdbCreateTSma(STsdb* pTsdb, char* pMsg);
int32_t tsdbInsertTSmaData(STsdb* pTsdb, int64_t indexUid, const char* msg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
uint64_t taskId);
uint64_t taskId);
tsdbReaderT tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
void* pMemRef);
int32_t tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo);
@ -189,7 +190,6 @@ struct STbUidStore {
#define TD_VID(PVNODE) (PVNODE)->config.vgId
static FORCE_INLINE bool vnodeIsRollup(SVnode* pVnode) {
SRetention* pRetention = &(pVnode->config.tsdbCfg.retentions[0]);
return (pRetention->freq > 0 && pRetention->keep > 0);

16
source/dnode/vnode/src/meta/metaTDBImpl.c
View File

@ -289,7 +289,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
pVal = pBuf = buf;
metaEncodeTbInfo(&pBuf, pTbCfg);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbPut(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, &pMetaDb->txn);
if (ret < 0) {
return -1;
}
@ -311,7 +311,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
pVal = pBuf = buf;
metaEncodeSchemaEx(&pBuf, &schemaWrapper);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbPut(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, &pMeta->pDB->txn);
ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, &pMeta->pDB->txn);
if (ret < 0) {
return -1;
}
@ -325,7 +325,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
kLen = nameLen + 1 + sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbPut(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
if (ret < 0) {
return -1;
}
@ -336,7 +336,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbPut(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
if (ret < 0) {
return -1;
}
@ -347,7 +347,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
kLen = sizeof(ctbIdxKey);
pVal = NULL;
vLen = 0;
ret = tdbDbPut(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
if (ret < 0) {
return -1;
}
@ -362,7 +362,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbPut(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
if (ret < 0) {
return -1;
}
@ -530,7 +530,7 @@ int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
int32_t kLen = sizeof(pSmaCfg->indexUid);
int32_t vLen = POINTER_DISTANCE(qBuf, pBuf);
ret = tdbDbPut(pMeta->pDB->pSmaDB, key, kLen, val, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMeta->pDB->pSmaDB, key, kLen, val, vLen, &pMetaDb->txn);
if (ret < 0) {
taosMemoryFreeClear(pBuf);
return -1;
@ -545,7 +545,7 @@ int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
val = NULL;
vLen = 0;
ret = tdbDbPut(pMeta->pDB->pSmaIdx, key, kLen, val, vLen, &pMetaDb->txn);
ret = tdbDbInsert(pMeta->pDB->pSmaIdx, key, kLen, val, vLen, &pMetaDb->txn);
if (ret < 0) {
taosMemoryFreeClear(pBuf);
return -1;

205
source/dnode/vnode/src/meta/metaTable.c
View File

@ -72,44 +72,61 @@ _err:
}
int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
SMetaReader mr = {0};
TDBC *pNameIdxc = NULL;
TDBC *pUidIdxc = NULL;
TDBC *pCtbIdxc = NULL;
SCtbIdxKey *pCtbIdxKey;
const void *pKey = NULL;
int nKey;
const void *pData = NULL;
int nData;
int c, ret;
// validate req
metaReaderInit(&mr, pMeta, 0);
if (metaGetTableEntryByUid(&mr, pReq->suid) < 0) {
terrno = TSDB_CODE_VND_TABLE_NOT_EXIST;
// prepare uid idx cursor
tdbDbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pUidIdxc, &pReq->suid, sizeof(tb_uid_t), &c);
if (ret < 0 || c != 0) {
terrno = TSDB_CODE_VND_TB_NOT_EXIST;
tdbDbcClose(pUidIdxc);
goto _err;
}
// do drop
// drop from pTbDb
// drop from pSkmDb
// drop from pUidIdx
// drop from pNameIdx
// {
// TDBC *pDbc1 = NULL;
// void *pKey = NULL;
// void *pVal = NULL;
// int kLen = 0;
// int vLen = 0;
// int ret = 0;
// prepare name idx cursor
tdbDbcOpen(pMeta->pNameIdx, &pNameIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pNameIdxc, pReq->name, strlen(pReq->name) + 1, &c);
if (ret < 0 || c != 0) {
ASSERT(0);
}
// // drop from pCtbIdx
// ret = tdbDbcOpen(pMeta->pCtbIdx, &pDbc1);
// tdbDbcMoveTo(pDbc1, &pReq->suid, sizeof(pReq->suid), NULL /*cmpr*/, 0 /*TDB_FORWARD_SEARCH*/);
// tdbDbcGet(pDbc1, &pKey, &kLen, &pVal, vLen);
// tdbDbcDrop(pDbc1);
// // drop from pTagIdx
// // drop from pTtlIdx
// }
tdbDbcDelete(pUidIdxc);
tdbDbcDelete(pNameIdxc);
tdbDbcClose(pUidIdxc);
tdbDbcClose(pNameIdxc);
// clear and return
metaReaderClear(&mr);
metaError("vgId:%d super table %s uid:%" PRId64 " is dropped", TD_VID(pMeta->pVnode), pReq->name, pReq->suid);
// loop to drop each child table
tdbDbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pCtbIdxc, &(SCtbIdxKey){.suid = pReq->suid, .uid = INT64_MIN}, sizeof(SCtbIdxKey), &c);
if (ret < 0 || (c < 0 && tdbDbcMoveToNext(pCtbIdxc) < 0)) {
tdbDbcClose(pCtbIdxc);
goto _exit;
}
for (;;) {
tdbDbcGet(pCtbIdxc, &pKey, &nKey, NULL, NULL);
pCtbIdxKey = (SCtbIdxKey *)pKey;
if (pCtbIdxKey->suid > pReq->suid) break;
// drop the child table (TODO)
if (tdbDbcMoveToNext(pCtbIdxc) < 0) break;
}
_exit:
metaDebug("vgId:%d super table %s uid:%" PRId64 " is dropped", TD_VID(pMeta->pVnode), pReq->name, pReq->suid);
return 0;
_err:
metaReaderClear(&mr);
metaError("vgId:%d failed to drop super table %s uid:%" PRId64 " since %s", TD_VID(pMeta->pVnode), pReq->name,
pReq->suid, tstrerror(terrno));
return -1;
@ -166,18 +183,122 @@ _err:
return -1;
}
int metaDropTable(SMeta *pMeta, tb_uid_t uid) {
#if 0
if (metaRemoveTableFromIdx(pMeta, uid) < 0) {
// TODO: handle error
int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
TDBC *pTbDbc = NULL;
TDBC *pUidIdxc = NULL;
TDBC *pNameIdxc = NULL;
const void *pData;
int nData;
tb_uid_t uid;
int64_t tver;
SMetaEntry me = {0};
SCoder coder = {0};
int8_t type;
int64_t ctime;
tb_uid_t suid;
int c, ret;
// search & delete the name idx
tdbDbcOpen(pMeta->pNameIdx, &pNameIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pNameIdxc, pReq->name, strlen(pReq->name) + 1, &c);
if (ret < 0 || c) {
tdbDbcClose(pNameIdxc);
terrno = TSDB_CODE_VND_TABLE_NOT_EXIST;
return -1;
}
if (metaRemoveTableFromIdx(pMeta, uid) < 0) {
// TODO
ret = tdbDbcGet(pNameIdxc, NULL, NULL, &pData, &nData);
if (ret < 0) {
ASSERT(0);
return -1;
}
#endif
uid = *(tb_uid_t *)pData;
tdbDbcDelete(pNameIdxc);
tdbDbcClose(pNameIdxc);
// search & delete uid idx
tdbDbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pUidIdxc, &uid, sizeof(uid), &c);
if (ret < 0 || c != 0) {
ASSERT(0);
return -1;
}
ret = tdbDbcGet(pUidIdxc, NULL, NULL, &pData, &nData);
if (ret < 0) {
ASSERT(0);
return -1;
}
tver = *(int64_t *)pData;
tdbDbcDelete(pUidIdxc);
tdbDbcClose(pUidIdxc);
// search and get meta entry
tdbDbcOpen(pMeta->pTbDb, &pTbDbc, &pMeta->txn);
ret = tdbDbcMoveTo(pTbDbc, &(STbDbKey){.uid = uid, .version = tver}, sizeof(STbDbKey), &c);
if (ret < 0 || c != 0) {
ASSERT(0);
return -1;
}
ret = tdbDbcGet(pTbDbc, NULL, NULL, &pData, &nData);
if (ret < 0) {
ASSERT(0);
return -1;
}
// decode entry
void *pDataCopy = taosMemoryMalloc(nData); // remove the copy (todo)
memcpy(pDataCopy, pData, nData);
tCoderInit(&coder, TD_LITTLE_ENDIAN, pDataCopy, nData, TD_DECODER);
ret = metaDecodeEntry(&coder, &me);
if (ret < 0) {
ASSERT(0);
return -1;
}
type = me.type;
if (type == TSDB_CHILD_TABLE) {
ctime = me.ctbEntry.ctime;
suid = me.ctbEntry.suid;
} else if (type == TSDB_NORMAL_TABLE) {
ctime = me.ntbEntry.ctime;
suid = 0;
} else {
ASSERT(0);
}
taosMemoryFree(pDataCopy);
tCoderClear(&coder);
tdbDbcClose(pTbDbc);
if (type == TSDB_CHILD_TABLE) {
// remove the pCtbIdx
TDBC *pCtbIdxc = NULL;
tdbDbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
ret = tdbDbcMoveTo(pCtbIdxc, &(SCtbIdxKey){.suid = suid, .uid = uid}, sizeof(SCtbIdxKey), &c);
if (ret < 0 || c != 0) {
ASSERT(0);
return -1;
}
tdbDbcDelete(pCtbIdxc);
tdbDbcClose(pCtbIdxc);
// remove tags from pTagIdx (todo)
} else if (type == TSDB_NORMAL_TABLE) {
// remove from pSkmDb
} else {
ASSERT(0);
}
// remove from ttl (todo)
if (ctime > 0) {
}
return 0;
}
@ -218,7 +339,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
tCoderClear(&coder);
// write to table.db
if (tdbDbPut(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
if (tdbDbInsert(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
goto _err;
}
@ -231,11 +352,11 @@ _err:
}
static int metaUpdateUidIdx(SMeta *pMeta, const SMetaEntry *pME) {
return tdbDbPut(pMeta->pUidIdx, &pME->uid, sizeof(tb_uid_t), &pME->version, sizeof(int64_t), &pMeta->txn);
return tdbDbInsert(pMeta->pUidIdx, &pME->uid, sizeof(tb_uid_t), &pME->version, sizeof(int64_t), &pMeta->txn);
}
static int metaUpdateNameIdx(SMeta *pMeta, const SMetaEntry *pME) {
return tdbDbPut(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), &pMeta->txn);
return tdbDbInsert(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), &pMeta->txn);
}
static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
@ -258,12 +379,12 @@ static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
ttlKey.dtime = ctime + ttlDays * 24 * 60 * 60;
ttlKey.uid = pME->uid;
return tdbDbPut(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), NULL, 0, &pMeta->txn);
return tdbDbInsert(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), NULL, 0, &pMeta->txn);
}
static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) {
SCtbIdxKey ctbIdxKey = {.suid = pME->ctbEntry.suid, .uid = pME->uid};
return tdbDbPut(pMeta->pCtbIdx, &ctbIdxKey, sizeof(ctbIdxKey), NULL, 0, &pMeta->txn);
return tdbDbInsert(pMeta->pCtbIdx, &ctbIdxKey, sizeof(ctbIdxKey), NULL, 0, &pMeta->txn);
}
static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) {
@ -304,7 +425,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER);
tEncodeSSchemaWrapper(&coder, pSW);
if (tdbDbPut(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
if (tdbDbInsert(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
rcode = -1;
goto _exit;
}

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

@ -97,7 +97,7 @@ int32_t tsdbCloseDBF(SDBFile *pDBF) {
int32_t tsdbSaveSmaToDB(SDBFile *pDBF, void *pKey, int32_t keyLen, void *pVal, int32_t valLen, TXN *txn) {
int32_t ret;
ret = tdbDbPut(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
ret = tdbDbInsert(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
if (ret < 0) {
tsdbError("Failed to create insert sma data into db, ret = %d", ret);
return -1;

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

@ -445,14 +445,45 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
}
static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
SVDropTbReq req = {0};
SVDropTbReq rsp = {0};
SVDropTbBatchReq req = {0};
SVDropTbBatchRsp rsp = {0};
SCoder coder = {0};
int ret;
pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
pRsp->pCont = NULL;
pRsp->contLen = 0;
pRsp->code = TSDB_CODE_SUCCESS;
// decode req
tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
ret = tDecodeSVDropTbBatchReq(&coder, &req);
if (ret < 0) {
terrno = TSDB_CODE_INVALID_MSG;
pRsp->code = terrno;
goto _exit;
}
// process req
rsp.pArray = taosArrayInit(sizeof(SVDropTbRsp), req.nReqs);
for (int iReq = 0; iReq < req.nReqs; iReq++) {
SVDropTbReq *pDropTbReq = req.pReqs + iReq;
SVDropTbRsp dropTbRsp = {0};
// return rsp
/* code */
ret = metaDropTable(pVnode->pMeta, version, pDropTbReq);
if (ret < 0) {
dropTbRsp.code = TSDB_CODE_SUCCESS;
} else {
dropTbRsp.code = terrno;
}
taosArrayPush(rsp.pArray, &dropTbRsp);
}
_exit:
tCoderClear(&coder);
// encode rsp (TODO)
return 0;
}
@ -482,7 +513,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
}
int32_t tsdbProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
if(!pReq) {
if (!pReq) {
terrno = TSDB_CODE_INVALID_PTR;
return TSDB_CODE_FAILED;
}

35
source/libs/executor/src/executorimpl.c
View File

@ -4830,6 +4830,7 @@ static SArray* createSortInfo(SNodeList* pNodeList, SNodeList* pNodeListTarget);
static SArray* createIndexMap(SNodeList* pNodeList);
static SArray* extractPartitionColInfo(SNodeList* pNodeList);
static int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
static void setJoinColumnInfo(SColumnInfo* pInfo, const SColumnNode* pLeftNode);
static SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
SInterval interval = {
@ -5624,25 +5625,29 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOf
goto _error;
}
pOperator->resultInfo.capacity = 4096;
pOperator->resultInfo.threshold = 4096 * 0.75;
initResultSizeInfo(pOperator, 4096);
// initResultRowInf
// o(&pInfo->binfo.resultRowInfo, 8);
pInfo->pRes = pResBlock;
pOperator->name = "JoinOperator";
pInfo->pRes = pResBlock;
pOperator->name = "MergeJoinOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_JOIN;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols;
pOperator->info = pInfo;
pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo;
SOperatorNode* pNode = (SOperatorNode*)pOnCondition;
setJoinColumnInfo(&pInfo->leftCol, (SColumnNode*)pNode->pLeft);
setJoinColumnInfo(&pInfo->rightCol, (SColumnNode*)pNode->pRight);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL);
int32_t code = appendDownstream(pOperator, pDownstream, numOfDownstream);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
return pOperator;
_error:
@ -5651,3 +5656,11 @@ _error:
pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) {
pColumn->slotId = pColumnNode->slotId;
pColumn->type = pColumnNode->node.resType.type;
pColumn->bytes = pColumnNode->node.resType.bytes;
pColumn->precision = pColumnNode->node.resType.precision;
pColumn->scale = pColumnNode->node.resType.scale;
}

14
source/libs/executor/src/indexoperator.c
View File

@ -63,9 +63,14 @@ typedef struct SIFParam {
} SIFParam;
static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
if (src == OP_TYPE_GREATER_THAN || src == OP_TYPE_GREATER_EQUAL || src == OP_TYPE_LOWER_THAN ||
src == OP_TYPE_LOWER_EQUAL) {
*dst = QUERY_RANGE;
if (src == OP_TYPE_GREATER_THAN) {
*dst = QUERY_GREATER_THAN;
} else if (src == OP_TYPE_GREATER_EQUAL) {
*dst = QUERY_GREATER_EQUAL;
} else if (src == OP_TYPE_LOWER_THAN) {
*dst = QUERY_LESS_THAN;
} else if (src == OP_TYPE_LOWER_EQUAL) {
*dst = QUERY_LESS_EQUAL;
} else if (src == OP_TYPE_EQUAL) {
*dst = QUERY_TERM;
} else if (src == OP_TYPE_LIKE || src == OP_TYPE_MATCH || src == OP_TYPE_NMATCH) {
@ -249,9 +254,6 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu
static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) {
SIndexTerm *tm = indexTermCreate(left->suid, DEFAULT, operType, left->colValType, left->colName,
strlen(left->colName), right->condValue, strlen(right->condValue));
if (operType == OP_TYPE_LOWER_EQUAL || operType == OP_TYPE_GREATER_EQUAL || operType == OP_TYPE_GREATER_THAN ||
operType == OP_TYPE_LOWER_THAN) {
}
if (tm == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}

7
source/libs/index/inc/indexFst.h
View File

@ -52,7 +52,6 @@ typedef struct FstRange {
uint64_t end;
} FstRange;
typedef enum { GE, GT, LE, LT } RangeType;
typedef enum { OneTransNext, OneTrans, AnyTrans, EmptyFinal } State;
typedef enum { Ordered, OutOfOrdered, DuplicateKey } OrderType;
@ -174,9 +173,9 @@ Output fstStateFinalOutput(FstState* state, uint64_t version, FstSlice* date,
uint64_t fstStateFindInput(FstState* state, FstNode* node, uint8_t b, bool* null);
#define FST_STATE_ONE_TRNAS_NEXT(node) (node->state.state == OneTransNext)
#define FST_STATE_ONE_TRNAS(node) (node->state.state == OneTrans)
#define FST_STATE_ANY_TRANS(node) (node->state.state == AnyTrans)
#define FST_STATE_EMPTY_FINAL(node) (node->state.state == EmptyFinal)
#define FST_STATE_ONE_TRNAS(node) (node->state.state == OneTrans)
#define FST_STATE_ANY_TRANS(node) (node->state.state == AnyTrans)
#define FST_STATE_EMPTY_FINAL(node) (node->state.state == EmptyFinal)
typedef struct FstLastTransition {
uint8_t inp;

2
source/libs/index/inc/indexInt.h
View File

@ -34,6 +34,7 @@
extern "C" {
#endif
typedef enum { LT, LE, GT, GE } RangeType;
typedef enum { kTypeValue, kTypeDeletion } STermValueType;
typedef struct SIndexStat {
@ -86,7 +87,6 @@ typedef struct SIndexTerm {
int32_t nColName;
char* colVal;
int32_t nColVal;
int8_t qType; // just use for range
} SIndexTerm;
typedef struct SIndexTermQuery {

1
source/libs/index/src/index.c
View File

@ -262,7 +262,6 @@ SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn oper, int8_t queryT
tm->colVal = (char*)taosMemoryCalloc(1, nColVal + 1);
memcpy(tm->colVal, colVal, nColVal);
tm->nColVal = nColVal;
tm->qType = queryType;
return tm;
}

67
source/libs/index/src/indexCache.c
View File

@ -38,10 +38,29 @@ static int32_t cacheSearchTerm(void* cache, CacheTerm* ct, SIdxTempResult* tr, S
static int32_t cacheSearchPrefix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchSuffix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchRegex(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchLessThan(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchLessEqual(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchGreaterThan(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchGreaterEqual(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
static int32_t cacheSearchRange(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s);
/*comm func of compare, used in (LE/LT/GE/GT compare)*/
static int32_t cacheSearchCompareFunc(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s,
RangeType type);
typedef enum { MATCH, CONTINUE, BREAK } TExeCond;
typedef TExeCond (*_cache_range_compare)(void* a, void* b, int8_t type);
static TExeCond tCompareLessThan(void* a, void* b, int8_t type) { return MATCH; }
static TExeCond tCompareLessEqual(void* a, void* b, int8_t type) { return MATCH; }
static TExeCond tCompareGreaterThan(void* a, void* b, int8_t type) { return MATCH; }
static TExeCond tCompareGreaterEqual(void* a, void* b, int8_t type) { return MATCH; }
static TExeCond (*rangeCompare[])(void* a, void* b, int8_t type) = {tCompareLessThan, tCompareLessEqual,
tCompareGreaterThan, tCompareGreaterEqual};
static int32_t (*cacheSearch[])(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) = {
cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchRange};
cacheSearchTerm, cacheSearchPrefix, cacheSearchSuffix, cacheSearchRegex, cacheSearchLessThan,
cacheSearchLessEqual, cacheSearchGreaterThan, cacheSearchGreaterEqual, cacheSearchRange};
static void doMergeWork(SSchedMsg* msg);
static bool indexCacheIteratorNext(Iterate* itera);
@ -88,6 +107,52 @@ static int32_t cacheSearchRegex(void* cache, CacheTerm* ct, SIdxTempResult* tr,
// impl later
return 0;
}
static int32_t cacheSearchCompareFunc(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s,
RangeType type) {
if (cache == NULL) {
return 0;
}
_cache_range_compare cmpFn = rangeCompare[type];
MemTable* mem = cache;
char* key = indexCacheTermGet(ct);
SSkipListIterator* iter = tSkipListCreateIter(mem->mem);
while (tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
if (node == NULL) {
break;
}
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
TExeCond cond = cmpFn(c->colVal, ct->colVal, ct->colType);
if (cond == MATCH) {
if (c->operaType == ADD_VALUE) {
INDEX_MERGE_ADD_DEL(tr->deled, tr->added, c->uid)
// taosArrayPush(result, &c->uid);
*s = kTypeValue;
} else if (c->operaType == DEL_VALUE) {
INDEX_MERGE_ADD_DEL(tr->added, tr->deled, c->uid)
}
} else if (cond == CONTINUE) {
} else if (cond == BREAK) {
break;
}
}
tSkipListDestroyIter(iter);
return TSDB_CODE_SUCCESS;
}
static int32_t cacheSearchLessThan(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, ct, tr, s, LT);
}
static int32_t cacheSearchLessEqual(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, ct, tr, s, LE);
}
static int32_t cacheSearchGreaterThan(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, ct, tr, s, GT);
}
static int32_t cacheSearchGreaterEqual(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
return cacheSearchCompareFunc(cache, ct, tr, s, GE);
}
static int32_t cacheSearchRange(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
// impl later
return 0;

50
source/libs/index/src/indexTfile.c
View File

@ -64,10 +64,17 @@ static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr);
static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType ctype);
static int32_t (*tfSearch[])(void* reader, SIndexTerm* tem, SIdxTempResult* tr) = {
tfSearchTerm, tfSearchPrefix, tfSearchSuffix, tfSearchRegex, tfSearchRange};
tfSearchTerm, tfSearchPrefix, tfSearchSuffix, tfSearchRegex, tfSearchLessThan,
tfSearchLessEqual, tfSearchGreaterThan, tfSearchGreaterEqual, tfSearchRange};
TFileCache* tfileCacheCreate(const char* path) {
TFileCache* tcache = taosMemoryCalloc(1, sizeof(TFileCache));
@ -299,6 +306,47 @@ static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr)
fstSliceDestroy(&key);
return 0;
}
static int32_t tfSearchCompareFunc(void* reader, SIndexTerm* tem, SIdxTempResult* tr, RangeType type) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0;
char* p = tem->colVal;
uint64_t sz = tem->nColVal;
if (hasJson) {
p = indexPackJsonData(tem);
sz = strlen(p);
}
SArray* offsets = taosArrayInit(16, sizeof(uint64_t));
AutomationCtx* ctx = automCtxCreate((void*)p, AUTOMATION_ALWAYS);
FstStreamBuilder* sb = fstSearch(((TFileReader*)reader)->fst, ctx);
FstSlice h = fstSliceCreate((uint8_t*)p, sz);
fstStreamBuilderSetRange(sb, &h, type);
fstSliceDestroy(&h);
StreamWithState* st = streamBuilderIntoStream(sb);
StreamWithStateResult* rt = NULL;
while ((rt = streamWithStateNextWith(st, NULL)) != NULL) {
taosArrayPush(offsets, &(rt->out.out));
swsResultDestroy(rt);
}
streamWithStateDestroy(st);
fstStreamBuilderDestroy(sb);
return TSDB_CODE_SUCCESS;
}
static int32_t tfSearchLessThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
return tfSearchCompareFunc(reader, tem, tr, LT);
}
static int32_t tfSearchLessEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
return tfSearchCompareFunc(reader, tem, tr, LE);
}
static int32_t tfSearchGreaterThan(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
return tfSearchCompareFunc(reader, tem, tr, GT);
}
static int32_t tfSearchGreaterEqual(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
return tfSearchCompareFunc(reader, tem, tr, GE);
}
static int32_t tfSearchRange(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
int ret = 0;

22
source/libs/scalar/src/sclvector.c
View File

@ -297,6 +297,22 @@ static FORCE_INLINE void varToNchar(char* buf, SScalarParam* pOut, int32_t rowIn
taosMemoryFree(t);
}
static FORCE_INLINE void ncharToVar(char* buf, SScalarParam* pOut, int32_t rowIndex) {
int32_t inputLen = varDataLen(buf);
char* t = taosMemoryCalloc(1, inputLen + VARSTR_HEADER_SIZE);
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(buf), varDataLen(buf), varDataVal(t));
if (len < 0) {
taosMemoryFree(t);
return;
}
varDataSetLen(t, len);
colDataAppend(pOut->columnData, rowIndex, t, false);
taosMemoryFree(t);
}
//TODO opt performance, tmp is not needed.
int32_t vectorConvertFromVarData(const SScalarParam* pIn, SScalarParam* pOut, int32_t inType, int32_t outType) {
int32_t bufSize = pIn->columnData->info.bytes;
@ -313,6 +329,10 @@ int32_t vectorConvertFromVarData(const SScalarParam* pIn, SScalarParam* pOut, in
func = varToUnsigned;
} else if (IS_FLOAT_TYPE(outType)) {
func = varToFloat;
} else if (outType == TSDB_DATA_TYPE_BINARY) { // nchar -> binary
ASSERT(inType == TSDB_DATA_TYPE_NCHAR);
func = ncharToVar;
vton = true;
} else if (outType == TSDB_DATA_TYPE_NCHAR) { // binary -> nchar
ASSERT(inType == TSDB_DATA_TYPE_VARCHAR);
func = varToNchar;
@ -608,7 +628,7 @@ int8_t gConvertTypes[TSDB_DATA_TYPE_BLOB+1][TSDB_DATA_TYPE_BLOB+1] = {
/*BIGI*/ 0, 0, 0, 0, 0, 0, 6, 7, 7, 0, 7, 5, 5, 5, 7, 0, 7, 0, 0,
/*FLOA*/ 0, 0, 0, 0, 0, 0, 0, 7, 7, 6, 7, 6, 6, 6, 6, 0, 7, 0, 0,
/*DOUB*/ 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 7, 7, 7, 0, 7, 0, 0,
/*VARC*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 0, 7, 7, 7, 7, 0, 0, 0, 0,
/*VARC*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 8, 7, 7, 7, 7, 0, 0, 0, 0,
/*TIME*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 9, 9, 9, 7, 0, 7, 0, 0,
/*NCHA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 0, 0, 0, 0,
/*UTIN*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 12, 13, 14, 0, 7, 0, 0,

10
source/libs/tdb/inc/tdb.h
View File

@ -40,7 +40,9 @@ int tdbCommit(TENV *pEnv, TXN *pTxn);
int tdbDbOpen(const char *fname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TENV *pEnv, TDB **ppDb);
int tdbDbClose(TDB *pDb);
int tdbDbDrop(TDB *pDb);
int tdbDbPut(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn);
int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn);
int tdbDbDelete(TDB *pDb, const void *pKey, int kLen, TXN *pTxn);
int tdbDbUpsert(TDB *pDb, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn);
int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen);
int tdbDbPGet(TDB *pDb, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
@ -53,11 +55,9 @@ int tdbDbcMoveToLast(TDBC *pDbc);
int tdbDbcMoveToNext(TDBC *pDbc);
int tdbDbcMoveToPrev(TDBC *pDbc);
int tdbDbcGet(TDBC *pDbc, const void **ppKey, int *pkLen, const void **ppVal, int *pvLen);
int tdbDbcPut(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen);
int tdbDbcUpdate(TDBC *pDbc, const void *pKey, int kLen, const void *pVal, int vLen);
int tdbDbcDrop(TDBC *pDbc);
int tdbDbcDelete(TDBC *pDbc);
int tdbDbcNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbDbcUpsert(TDBC *pDbc, const void *pKey, int nKey, const void *pData, int nData, int insert);
// TXN
#define TDB_TXN_WRITE 0x1

268
source/libs/tdb/src/db/tdbBtree.c
View File

@ -138,67 +138,90 @@ int tdbBtreeInsert(SBTree *pBt, const void *pKey, int kLen, const void *pVal, in
}
if (btc.idx == -1) {
idx = 0;
btc.idx = 0;
} else {
if (c > 0) {
idx = btc.idx + 1;
} else if (c < 0) {
idx = btc.idx;
} else {
// TDB does NOT allow same key
tdbBtcClose(&btc);
btc.idx++;
} else if (c == 0) {
// dup key not allowed
ASSERT(0);
return -1;
}
}
// make sure enough space to hold the cell
szBuf = kLen + vLen + 14;
pBuf = tdbRealloc(pBt->pBuf, pBt->pageSize > szBuf ? szBuf : pBt->pageSize);
if (pBuf == NULL) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
pBt->pBuf = pBuf;
pCell = (SCell *)pBt->pBuf;
// encode cell
ret = tdbBtreeEncodeCell(btc.pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
ret = tdbBtcUpsert(&btc, pKey, kLen, pVal, vLen, 1);
if (ret < 0) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
// mark the page dirty
ret = tdbPagerWrite(pBt->pPager, btc.pPage);
if (ret < 0) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
// insert the cell
ret = tdbPageInsertCell(btc.pPage, idx, pCell, szCell, 0);
if (ret < 0) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
// check if need balance
if (btc.pPage->nOverflow > 0) {
ret = tdbBtreeBalance(&btc);
if (ret < 0) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
}
tdbBtcClose(&btc);
return 0;
}
int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn) {
SBTC btc;
int c;
int ret;
tdbBtcOpen(&btc, pBt, pTxn);
// move the cursor
ret = tdbBtcMoveTo(&btc, pKey, kLen, &c);
if (ret < 0) {
tdbBtcClose(&btc);
ASSERT(0);
return -1;
}
if (btc.idx < 0 || c != 0) {
tdbBtcClose(&btc);
return -1;
}
// delete the key
if (tdbBtcDelete(&btc) < 0) {
tdbBtcClose(&btc);
return -1;
}
tdbBtcClose(&btc);
return 0;
}
int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn) {
SBTC btc;
int c;
int ret;
tdbBtcOpen(&btc, pBt, pTxn);
// move the cursor
ret = tdbBtcMoveTo(&btc, pKey, nKey, &c);
if (ret < 0) {
ASSERT(0);
tdbBtcClose(&btc);
return -1;
}
if (btc.idx == -1) {
btc.idx = 0;
c = 1;
} else {
if (c > 0) {
btc.idx = btc.idx + 1;
}
}
ret = tdbBtcUpsert(&btc, pKey, nKey, pData, nData, c);
if (ret < 0) {
ASSERT(0);
tdbBtcClose(&btc);
return -1;
}
tdbBtcClose(&btc);
return 0;
}
@ -552,14 +575,14 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx, TXN *pTx
SCell *pCell;
int szLCell, szRCell;
// balance page (iNew) and (iNew-1)
for (;;) {
pCell = tdbPageGetCell(pOlds[infoNews[iNew - 1].iPage], infoNews[iNew - 1].oIdx);
if (childNotLeaf) {
szLCell = szRCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
szLCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
if (!childNotLeaf) {
szRCell = szLCell;
} else {
szLCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
int iPage = infoNews[iNew - 1].iPage;
int oIdx = infoNews[iNew - 1].oIdx + 1;
SPage *pPage;
@ -736,6 +759,13 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx, TXN *pTx
}
}
if (TDB_BTREE_PAGE_IS_ROOT(pParent) && TDB_PAGE_TOTAL_CELLS(pParent) == 0) {
i8 flags = TDB_BTREE_ROOT | TDB_BTREE_PAGE_IS_LEAF(pNews[0]);
// copy content to the parent page
tdbBtreeInitPage(pParent, &(SBtreeInitPageArg){.flags = flags, .pBt = pBt}, 0);
tdbPageCopy(pNews[0], pParent);
}
for (int i = 0; i < 3; i++) {
if (pDivCell[i]) {
tdbOsFree(pDivCell[i]);
@ -1357,7 +1387,143 @@ int tdbBtcGet(SBTC *pBtc, const void **ppKey, int *kLen, const void **ppVal, int
if (ppVal) {
*ppVal = (void *)pBtc->coder.pVal;
*kLen = pBtc->coder.vLen;
*vLen = pBtc->coder.vLen;
}
return 0;
}
int tdbBtcDelete(SBTC *pBtc) {
int idx = pBtc->idx;
int nCells = TDB_PAGE_TOTAL_CELLS(pBtc->pPage);
SPager *pPager = pBtc->pBt->pPager;
const void *pKey;
i8 iPage;
SPage *pPage;
SPgno pgno;
SCell *pCell;
int szCell;
int nKey;
int ret;
ASSERT(idx >= 0 && idx < nCells);
// drop the cell on the leaf
ret = tdbPagerWrite(pPager, pBtc->pPage);
if (ret < 0) {
ASSERT(0);
return -1;
}
tdbPageDropCell(pBtc->pPage, idx);
// update interior page or do balance
if (idx == nCells - 1) {
if (idx) {
pBtc->idx--;
tdbBtcGet(pBtc, &pKey, &nKey, NULL, NULL);
// loop to update the interial page
pgno = TDB_PAGE_PGNO(pBtc->pPage);
for (iPage = pBtc->iPage - 1; iPage >= 0; iPage--) {
pPage = pBtc->pgStack[iPage];
idx = pBtc->idxStack[iPage];
nCells = TDB_PAGE_TOTAL_CELLS(pPage);
if (idx < nCells) {
ret = tdbPagerWrite(pPager, pPage);
if (ret < 0) {
ASSERT(0);
return -1;
}
// update the cell with new key
pCell = tdbOsMalloc(nKey + 9);
tdbBtreeEncodeCell(pPage, pKey, nKey, &pgno, sizeof(pgno), pCell, &szCell);
ret = tdbPageUpdateCell(pPage, idx, pCell, szCell);
if (ret < 0) {
tdbOsFree(pCell);
ASSERT(0);
return -1;
}
tdbOsFree(pCell);
break;
} else {
pgno = TDB_PAGE_PGNO(pPage);
}
}
} else {
// delete the leaf page and do balance
ASSERT(TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0);
ret = tdbBtreeBalance(pBtc);
if (ret < 0) {
ASSERT(0);
return -1;
}
}
}
return 0;
}
int tdbBtcUpsert(SBTC *pBtc, const void *pKey, int kLen, const void *pData, int nData, int insert) {
SCell *pCell;
int szCell;
int nCells = TDB_PAGE_TOTAL_CELLS(pBtc->pPage);
int szBuf;
void *pBuf;
int ret;
ASSERT(pBtc->idx >= 0);
// alloc space
szBuf = kLen + nData + 14;
pBuf = tdbRealloc(pBtc->pBt->pBuf, pBtc->pBt->pageSize > szBuf ? szBuf : pBtc->pBt->pageSize);
if (pBuf == NULL) {
ASSERT(0);
return -1;
}
pBtc->pBt->pBuf = pBuf;
pCell = (SCell *)pBtc->pBt->pBuf;
// encode cell
ret = tdbBtreeEncodeCell(pBtc->pPage, pKey, kLen, pData, nData, pCell, &szCell);
if (ret < 0) {
ASSERT(0);
return -1;
}
// mark dirty
ret = tdbPagerWrite(pBtc->pBt->pPager, pBtc->pPage);
if (ret < 0) {
ASSERT(0);
return -1;
}
// insert or update
if (insert) {
ASSERT(pBtc->idx <= nCells);
ret = tdbPageInsertCell(pBtc->pPage, pBtc->idx, pCell, szCell, 0);
} else {
ASSERT(pBtc->idx < nCells);
ret = tdbPageUpdateCell(pBtc->pPage, pBtc->idx, pCell, szCell);
}
if (ret < 0) {
ASSERT(0);
return -1;
}
// check balance
if (pBtc->pPage->nOverflow > 0) {
ret = tdbBtreeBalance(pBtc);
if (ret < 0) {
ASSERT(0);
return -1;
}
}
return 0;

30
source/libs/tdb/src/db/tdbDb.c
View File

@ -75,10 +75,16 @@ int tdbDbDrop(TDB *pDb) {
return 0;
}
int tdbDbPut(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn) {
int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn) {
return tdbBtreeInsert(pDb->pBt, pKey, keyLen, pVal, valLen, pTxn);
}
int tdbDbDelete(TDB *pDb, const void *pKey, int kLen, TXN *pTxn) { return tdbBtreeDelete(pDb->pBt, pKey, kLen, pTxn); }
int tdbDbUpsert(TDB *pDb, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn) {
return tdbBtreeUpsert(pDb->pBt, pKey, kLen, pVal, vLen, pTxn);
}
int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen) {
return tdbBtreeGet(pDb->pBt, pKey, kLen, ppVal, vLen);
}
@ -117,28 +123,16 @@ int tdbDbcGet(TDBC *pDbc, const void **ppKey, int *pkLen, const void **ppVal, in
return tdbBtcGet(&pDbc->btc, ppKey, pkLen, ppVal, pvLen);
}
int tdbDbcPut(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen) {
// TODO
ASSERT(0);
return 0;
}
int tdbDbcUpdate(TDBC *pDbc, const void *pKey, int kLen, const void *pVal, int vLen) {
// TODO
ASSERT(0);
return 0;
}
int tdbDbcDrop(TDBC *pDbc) {
// TODO
ASSERT(0);
return 0;
}
int tdbDbcDelete(TDBC *pDbc) { return tdbBtcDelete(&pDbc->btc); }
int tdbDbcNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen) {
return tdbBtreeNext(&pDbc->btc, ppKey, kLen, ppVal, vLen);
}
int tdbDbcUpsert(TDBC *pDbc, const void *pKey, int nKey, const void *pData, int nData, int insert) {
return tdbBtcUpsert(&pDbc->btc, pKey, nKey, pData, nData, insert);
}
int tdbDbcClose(TDBC *pDbc) {
if (pDbc) {
tdbBtcClose(&pDbc->btc);

5
source/libs/tdb/src/db/tdbPage.c
View File

@ -171,6 +171,11 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl
return 0;
}
int tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
tdbPageDropCell(pPage, idx);
return tdbPageInsertCell(pPage, idx, pCell, szCell, 0);
}
int tdbPageDropCell(SPage *pPage, int idx) {
int lidx;
SCell *pCell;

5
source/libs/tdb/src/inc/tdbInt.h
View File

@ -128,6 +128,8 @@ struct SBTC {
int tdbBtreeOpen(int keyLen, int valLen, SPager *pFile, tdb_cmpr_fn_t kcmpr, SBTree **ppBt);
int tdbBtreeClose(SBTree *pBt);
int tdbBtreeInsert(SBTree *pBt, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn);
int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn);
int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn);
int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen);
int tdbBtreePGet(SBTree *pBt, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
@ -141,6 +143,8 @@ int tdbBtcMoveToNext(SBTC *pBtc);
int tdbBtcMoveToPrev(SBTC *pBtc);
int tdbBtreeNext(SBTC *pBtc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbBtcGet(SBTC *pBtc, const void **ppKey, int *kLen, const void **ppVal, int *vLen);
int tdbBtcDelete(SBTC *pBtc);
int tdbBtcUpsert(SBTC *pBtc, const void *pKey, int kLen, const void *pData, int nData, int insert);
// tdbPager.c ====================================
@ -278,6 +282,7 @@ void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
int tdbPageDropCell(SPage *pPage, int idx);
int tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell);
void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
int tdbPageCapacity(int pageSize, int amHdrSize);

180
source/libs/tdb/test/tdbTest.cpp
View File

@ -115,12 +115,12 @@ static int tDefaultKeyCmpr(const void *pKey1, int keyLen1, const void *pKey2, in
return cret;
}
TEST(tdb_test, simple_test) {
TEST(tdb_test, simple_insert1) {
int ret;
TENV *pEnv;
TDB *pDb;
tdb_cmpr_fn_t compFunc;
int nData = 10000000;
int nData = 1000000;
TXN txn;
taosRemoveDir("tdb");
@ -152,7 +152,7 @@ TEST(tdb_test, simple_test) {
for (int iData = 1; iData <= nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(val, "value%d", iData);
ret = tdbDbPut(pDb, key, strlen(key), val, strlen(val), &txn);
ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val), &txn);
GTEST_ASSERT_EQ(ret, 0);
// if pool is full, commit the transaction and start a new one
@ -202,6 +202,8 @@ TEST(tdb_test, simple_test) {
ret = tdbDbcOpen(pDb, &pDBC, NULL);
GTEST_ASSERT_EQ(ret, 0);
tdbDbcMoveToFirst(pDBC);
for (;;) {
ret = tdbDbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
if (ret < 0) break;
@ -233,7 +235,7 @@ TEST(tdb_test, simple_test) {
GTEST_ASSERT_EQ(ret, 0);
}
TEST(tdb_test, simple_test2) {
TEST(tdb_test, simple_insert2) {
int ret;
TENV *pEnv;
TDB *pDb;
@ -269,7 +271,7 @@ TEST(tdb_test, simple_test2) {
for (int iData = 1; iData <= nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(val, "value%d", iData);
ret = tdbDbPut(pDb, key, strlen(key), val, strlen(val), &txn);
ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val), &txn);
GTEST_ASSERT_EQ(ret, 0);
}
@ -283,13 +285,15 @@ TEST(tdb_test, simple_test2) {
ret = tdbDbcOpen(pDb, &pDBC, NULL);
GTEST_ASSERT_EQ(ret, 0);
tdbDbcMoveToFirst(pDBC);
for (;;) {
ret = tdbDbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
if (ret < 0) break;
std::cout.write((char *)pKey, kLen) /* << " " << kLen */ << " ";
std::cout.write((char *)pVal, vLen) /* << " " << vLen */;
std::cout << std::endl;
// std::cout.write((char *)pKey, kLen) /* << " " << kLen */ << " ";
// std::cout.write((char *)pVal, vLen) /* << " " << vLen */;
// std::cout << std::endl;
count++;
}
@ -316,4 +320,164 @@ TEST(tdb_test, simple_test2) {
// Close Env
ret = tdbEnvClose(pEnv);
GTEST_ASSERT_EQ(ret, 0);
}
TEST(tdb_test, simple_delete1) {
int ret;
TDB *pDb;
char key[128];
char data[128];
TXN txn;
TENV *pEnv;
SPoolMem *pPool;
void *pKey = NULL;
void *pData = NULL;
int nKey;
TDBC *pDbc;
int nData;
int nKV = 69;
taosRemoveDir("tdb");
pPool = openPool();
// open env
ret = tdbEnvOpen("tdb", 1024, 256, &pEnv);
GTEST_ASSERT_EQ(ret, 0);
// open database
ret = tdbDbOpen("db.db", -1, -1, tKeyCmpr, pEnv, &pDb);
GTEST_ASSERT_EQ(ret, 0);
tdbTxnOpen(&txn, 0, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
tdbBegin(pEnv, &txn);
// loop to insert batch data
for (int iData = 0; iData < nKV; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d", iData);
ret = tdbDbInsert(pDb, key, strlen(key), data, strlen(data), &txn);
GTEST_ASSERT_EQ(ret, 0);
}
// query the data
for (int iData = 0; iData < nKV; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d", iData);
ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
GTEST_ASSERT_EQ(ret, 0);
GTEST_ASSERT_EQ(memcmp(data, pData, nData), 0);
}
// loop to delete some data
for (int iData = nKV - 1; iData > 30; iData--) {
sprintf(key, "key%d", iData);
ret = tdbDbDelete(pDb, key, strlen(key), &txn);
GTEST_ASSERT_EQ(ret, 0);
}
// query the data
for (int iData = 0; iData < nKV; iData++) {
sprintf(key, "key%d", iData);
ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
if (iData <= 30) {
GTEST_ASSERT_EQ(ret, 0);
} else {
GTEST_ASSERT_EQ(ret, -1);
}
}
// loop to iterate the data
tdbDbcOpen(pDb, &pDbc, NULL);
ret = tdbDbcMoveToFirst(pDbc);
GTEST_ASSERT_EQ(ret, 0);
pKey = NULL;
pData = NULL;
for (;;) {
ret = tdbDbcNext(pDbc, &pKey, &nKey, &pData, &nData);
if (ret < 0) break;
std::cout.write((char *)pKey, nKey) /* << " " << kLen */ << " ";
std::cout.write((char *)pData, nData) /* << " " << vLen */;
std::cout << std::endl;
}
tdbDbcClose(pDbc);
tdbCommit(pEnv, &txn);
closePool(pPool);
tdbDbClose(pDb);
tdbEnvClose(pEnv);
}
TEST(tdb_test, simple_upsert1) {
int ret;
TENV *pEnv;
TDB *pDb;
int nData = 100000;
char key[64];
char data[64];
void *pData = NULL;
SPoolMem *pPool;
TXN txn;
taosRemoveDir("tdb");
// open env
ret = tdbEnvOpen("tdb", 4096, 64, &pEnv);
GTEST_ASSERT_EQ(ret, 0);
// open database
ret = tdbDbOpen("db.db", -1, -1, NULL, pEnv, &pDb);
GTEST_ASSERT_EQ(ret, 0);
pPool = openPool();
// insert some data
tdbTxnOpen(&txn, 0, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
tdbBegin(pEnv, &txn);
for (int iData = 0; iData < nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d", iData);
ret = tdbDbInsert(pDb, key, strlen(key), data, strlen(data), &txn);
GTEST_ASSERT_EQ(ret, 0);
}
// query the data
for (int iData = 0; iData < nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d", iData);
ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
GTEST_ASSERT_EQ(ret, 0);
GTEST_ASSERT_EQ(memcmp(pData, data, nData), 0);
}
// upsert some data
for (int iData = 0; iData < nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d-u", iData);
ret = tdbDbUpsert(pDb, key, strlen(key), data, strlen(data), &txn);
GTEST_ASSERT_EQ(ret, 0);
}
tdbCommit(pEnv, &txn);
// query the data
for (int iData = 0; iData < nData; iData++) {
sprintf(key, "key%d", iData);
sprintf(data, "data%d-u", iData);
ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
GTEST_ASSERT_EQ(ret, 0);
GTEST_ASSERT_EQ(memcmp(pData, data, nData), 0);
}
tdbDbClose(pDb);
tdbEnvClose(pEnv);
}

175
source/util/src/tskiplist2.c Normal file
View File

@ -0,0 +1,175 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tskiplist2.h"
struct SSLNode {
int8_t level;
SSLNode *forwards[];
};
struct SSkipList2 {
int8_t level;
uint32_t seed;
int32_t size;
const SSLCfg *pCfg;
SSLNode *pHead[];
};
static void *slMalloc(void *pPool, int32_t size);
static void slFree(void *pPool, void *p);
static int32_t slCmprFn(const void *pKey, int32_t nKey, const void *pData, int32_t nData);
const SSLCfg slDefaultCfg = {.maxLevel = SL_MAX_LEVEL,
.nKey = -1,
.nData = -1,
.cmprFn = slCmprFn,
.pPool = NULL,
.xMalloc = slMalloc,
.xFree = slFree};
int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl) {
SSkipList2 *pSl = NULL;
int32_t size;
*ppSl = NULL;
if (pCfg == NULL) pCfg = &slDefaultCfg;
// check config (TODO)
// malloc handle
size = sizeof(*pSl) + sizeof(SSLNode *) * pCfg->maxLevel * 2;
pSl = pCfg->xMalloc(pCfg->pPool, size);
if (pSl == NULL) {
return -1;
}
pSl->level = 0;
pSl->seed = taosRand();
pSl->size = 0;
pSl->pCfg = pCfg;
// init an empty skiplist
for (int32_t i = 0; i < pCfg->maxLevel * 2; i++) {
pSl->pHead[i] = NULL;
}
*ppSl = pSl;
return 0;
}
int32_t slClose(SSkipList2 *pSl) {
if (pSl) {
slClear(pSl);
if (pSl->pCfg->xFree) {
pSl->pCfg->xFree(pSl->pCfg->pPool, pSl);
}
}
return 0;
}
int32_t slClear(SSkipList2 *pSl) {
// loop to clear sl
for (;;) {
// (TODO)
}
// init sl (TODO)
return 0;
}
int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc) {
pSlc->pSl = pSl;
for (int i = 0; i < SL_MAX_LEVEL; i++) {
if (i < pSl->pCfg->maxLevel) {
} else {
pSlc->forwards[i] = NULL;
}
}
// TODO
return 0;
}
int32_t slcClose(SSLCursor *pSlc) {
// TODO
return 0;
}
int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey) {
// TODO
return 0;
}
int32_t slcMoveToNext(SSLCursor *pSlc) {
// TODO
return 0;
}
int32_t slcMoveToPrev(SSLCursor *pSlc) {
// TODO
return 0;
}
int32_t slcMoveToFirst(SSLCursor *pSlc) {
// TODO
return 0;
}
int32_t slcMoveToLast(SSLCursor *pSlc) {
// TODO
return 0;
}
int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData) {
// TODO
return 0;
}
int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData) {
// TODO
return 0;
}
int32_t slcDrop(SSLCursor *pSlc) {
// TODO
return 0;
}
static FORCE_INLINE void *slMalloc(void *pPool, int32_t size) { return taosMemoryMalloc(size); }
static FORCE_INLINE void slFree(void *pPool, void *p) { taosMemoryFree(p); }
static int32_t slCmprFn(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2) {
ASSERT(nKey1 >= 0 && nKey2 >= 0);
int32_t nKey = nKey1 > nKey2 ? nKey2 : nKey1;
int32_t c;
c = memcmp(pKey1, pKey2, nKey);
if (c == 0) {
if (nKey1 > nKey2) {
c = 1;
} else if (nKey1 < nKey2) {
c = -1;
}
}
return c;
}

1928
tests/script/api/batchprepare.c
View File

File diff suppressed because it is too large Load Diff

2
tools/taos-tools

@ -1 +1 @@
Subproject commit bf6c766986c61ff4fc80421fdea682a8fd4b5b32
Subproject commit 2f3dfddd4d9a869e706ba3cf98fb6d769404cd7c