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

Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/vnode_refact1

This commit is contained in:
Hongze Cheng 2022-05-05 01:42:47 +00:00
parent c734e2d9a1 e50e3a17fe
commit f2859e9c1d
24 changed files with 846 additions and 514 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/common/tglobal.h
View File

@ -121,6 +121,10 @@ extern char tsCompressor[];
extern int32_t tsDiskCfgNum;
extern SDiskCfg tsDiskCfg[];
// internal
extern int32_t tsTransPullupMs;
extern int32_t tsMaRebalanceMs;
#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile,

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

@ -17,6 +17,7 @@
#define _TD_INDEX_H_
#include "os.h"
#include "taoserror.h"
#include "tarray.h"
#ifdef __cplusplus
@ -41,11 +42,22 @@ typedef enum {
UPDATE_VALUE, // update index column value
ADD_INDEX, // add index on specify column
DROP_INDEX, // drop existed index
DROP_SATBLE // drop stable
DROP_SATBLE, // drop stable
DEFAULT // query
} SIndexOperOnColumn;
typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType;
typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX = 3, QUERY_RANGE = 4 } EIndexQueryType;
typedef enum { MUST = 0, SHOULD, NOT } EIndexOperatorType;
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
@ -166,8 +178,8 @@ void indexOptsDestroy(SIndexOpts* opts);
* @param:
*/
SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType, const char* colName,
int32_t nColName, const char* colVal, int32_t nColVal);
SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn operType, int8_t qType, uint8_t colType,
const char* colName, int32_t nColName, const char* colVal, int32_t nColVal);
void indexTermDestroy(SIndexTerm* p);
/*

3
include/util/taoserror.h
View File

@ -264,7 +264,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_TRANS_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03D0)
#define TSDB_CODE_MND_TRANS_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03D1)
#define TSDB_CODE_MND_TRANS_INVALID_STAGE TAOS_DEF_ERROR_CODE(0, 0x03D2)
#define TSDB_CODE_MND_TRANS_CANT_PARALLEL TAOS_DEF_ERROR_CODE(0, 0x03D4)
#define TSDB_CODE_MND_TRANS_CONFLICT TAOS_DEF_ERROR_CODE(0, 0x03D3)
#define TSDB_CODE_MND_TRANS_UNKNOW_ERROR TAOS_DEF_ERROR_CODE(0, 0x03D4)
// mnode-mq
#define TSDB_CODE_MND_TOPIC_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03E0)

4
source/common/src/tglobal.c
View File

@ -169,6 +169,10 @@ uint32_t tsMaxRange = 500; // max range
uint32_t tsCurRange = 100; // range
char tsCompressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
// internal
int32_t tsTransPullupMs = 6000;
int32_t tsMaRebalanceMs = 2000;
void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) {
tstrncpy(tsDiskCfg[index].dir, v1, TSDB_FILENAME_LEN);
tsDiskCfg[index].level = level;

62
source/common/src/tmsgcb.c
View File

@ -15,37 +15,83 @@
#define _DEFAULT_SOURCE
#include "tmsgcb.h"
#include "taoserror.h"
static SMsgCb tsDefaultMsgCb;
void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb) { tsDefaultMsgCb = *pMsgCb; }
int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq) {
return (*pMsgCb->queueFps[qtype])(pMsgCb->pWrapper, pReq);
PutToQueueFp fp = pMsgCb->queueFps[qtype];
if (fp != NULL) {
return (*fp)(pMsgCb->pWrapper, pReq);
} else {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
}
int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype) {
return (*pMsgCb->qsizeFp)(pMsgCb->pWrapper, vgId, qtype);
GetQueueSizeFp fp = pMsgCb->qsizeFp;
if (fp != NULL) {
return (*fp)(pMsgCb->pWrapper, vgId, qtype);
} else {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
}
int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq) {
return (*pMsgCb->sendReqFp)(pMsgCb->pWrapper, epSet, pReq);
SendReqFp fp = pMsgCb->sendReqFp;
if (fp != NULL) {
return (*fp)(pMsgCb->pWrapper, epSet, pReq);
} else {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
}
void tmsgSendRsp(const SRpcMsg* pRsp) { return (*tsDefaultMsgCb.sendRspFp)(tsDefaultMsgCb.pWrapper, pRsp); }
void tmsgSendRsp(const SRpcMsg* pRsp) {
SendRspFp fp = tsDefaultMsgCb.sendRspFp;
if (fp != NULL) {
return (*fp)(tsDefaultMsgCb.pWrapper, pRsp);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}
void tmsgSendRedirectRsp(const SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
return (*tsDefaultMsgCb.sendRedirectRspFp)(tsDefaultMsgCb.pWrapper, pRsp, pNewEpSet);
SendRedirectRspFp fp = tsDefaultMsgCb.sendRedirectRspFp;
if (fp != NULL) {
(*fp)(tsDefaultMsgCb.pWrapper, pRsp, pNewEpSet);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}
void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg) {
(*pMsgCb->registerBrokenLinkArgFp)(pMsgCb->pWrapper, pMsg);
RegisterBrokenLinkArgFp fp = pMsgCb->registerBrokenLinkArgFp;
if (fp != NULL) {
(*fp)(pMsgCb->pWrapper, pMsg);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}
void tmsgReleaseHandle(void* handle, int8_t type) {
(*tsDefaultMsgCb.releaseHandleFp)(tsDefaultMsgCb.pWrapper, handle, type);
ReleaseHandleFp fp = tsDefaultMsgCb.releaseHandleFp;
if (fp != NULL) {
(*fp)(tsDefaultMsgCb.pWrapper, handle, type);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}
void tmsgReportStartup(const char* name, const char* desc) {
(*tsDefaultMsgCb.reportStartupFp)(tsDefaultMsgCb.pWrapper, name, desc);
ReportStartup fp = tsDefaultMsgCb.reportStartupFp;
if (fp != NULL && tsDefaultMsgCb.pWrapper != NULL) {
(*fp)(tsDefaultMsgCb.pWrapper, name, desc);
} else {
terrno = TSDB_CODE_INVALID_PTR;
}
}

16
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -57,11 +57,11 @@ typedef enum {
TRN_STAGE_PREPARE = 0,
TRN_STAGE_REDO_LOG = 1,
TRN_STAGE_REDO_ACTION = 2,
TRN_STAGE_COMMIT = 3,
TRN_STAGE_COMMIT_LOG = 4,
TRN_STAGE_UNDO_ACTION = 5,
TRN_STAGE_UNDO_LOG = 6,
TRN_STAGE_ROLLBACK = 7,
TRN_STAGE_ROLLBACK = 3,
TRN_STAGE_UNDO_ACTION = 4,
TRN_STAGE_UNDO_LOG = 5,
TRN_STAGE_COMMIT = 6,
TRN_STAGE_COMMIT_LOG = 7,
TRN_STAGE_FINISHED = 8
} ETrnStage;
@ -72,6 +72,7 @@ typedef enum {
TRN_TYPE_DROP_USER = 1003,
TRN_TYPE_CREATE_FUNC = 1004,
TRN_TYPE_DROP_FUNC = 1005,
TRN_TYPE_CREATE_SNODE = 1006,
TRN_TYPE_DROP_SNODE = 1007,
TRN_TYPE_CREATE_QNODE = 1008,
@ -91,10 +92,12 @@ typedef enum {
TRN_TYPE_CONSUMER_LOST = 1022,
TRN_TYPE_CONSUMER_RECOVER = 1023,
TRN_TYPE_BASIC_SCOPE_END,
TRN_TYPE_GLOBAL_SCOPE = 2000,
TRN_TYPE_CREATE_DNODE = 2001,
TRN_TYPE_DROP_DNODE = 2002,
TRN_TYPE_GLOBAL_SCOPE_END,
TRN_TYPE_DB_SCOPE = 3000,
TRN_TYPE_CREATE_DB = 3001,
TRN_TYPE_ALTER_DB = 3002,
@ -102,6 +105,7 @@ typedef enum {
TRN_TYPE_SPLIT_VGROUP = 3004,
TRN_TYPE_MERGE_VGROUP = 3015,
TRN_TYPE_DB_SCOPE_END,
TRN_TYPE_STB_SCOPE = 4000,
TRN_TYPE_CREATE_STB = 4001,
TRN_TYPE_ALTER_STB = 4002,
@ -131,7 +135,7 @@ typedef struct {
int32_t id;
ETrnStage stage;
ETrnPolicy policy;
ETrnType transType;
ETrnType type;
int32_t code;
int32_t failedTimes;
void* rpcHandle;

223
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -63,13 +63,15 @@ static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
int32_t mndInitTrans(SMnode *pMnode) {
SSdbTable table = {.sdbType = SDB_TRANS,
.keyType = SDB_KEY_INT32,
.encodeFp = (SdbEncodeFp)mndTransActionEncode,
.decodeFp = (SdbDecodeFp)mndTransActionDecode,
.insertFp = (SdbInsertFp)mndTransActionInsert,
.updateFp = (SdbUpdateFp)mndTransActionUpdate,
.deleteFp = (SdbDeleteFp)mndTransActionDelete};
SSdbTable table = {
.sdbType = SDB_TRANS,
.keyType = SDB_KEY_INT32,
.encodeFp = (SdbEncodeFp)mndTransActionEncode,
.decodeFp = (SdbDecodeFp)mndTransActionDecode,
.insertFp = (SdbInsertFp)mndTransActionInsert,
.updateFp = (SdbUpdateFp)mndTransActionUpdate,
.deleteFp = (SdbDeleteFp)mndTransActionDelete,
};
mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransReq);
mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
@ -123,71 +125,82 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
}
int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, pTrans->id, TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->policy, TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->stage, TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->transType, TRANS_ENCODE_OVER)
SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, TRANS_ENCODE_OVER)
SDB_SET_INT64(pRaw, dataPos, pTrans->dbUid, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_DB_FNAME_LEN, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, redoLogNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, undoLogNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, commitLogNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, redoActionNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, undoActionNum, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
ETrnStage stage = pTrans->stage;
if (stage == TRN_STAGE_REDO_LOG || stage == TRN_STAGE_REDO_ACTION) {
stage = TRN_STAGE_PREPARE;
} else if (stage == TRN_STAGE_UNDO_ACTION || stage == TRN_STAGE_UNDO_LOG) {
stage = TRN_STAGE_ROLLBACK;
} else if (stage == TRN_STAGE_COMMIT_LOG || stage == TRN_STAGE_FINISHED) {
stage = TRN_STAGE_COMMIT;
} else {
}
SDB_SET_INT16(pRaw, dataPos, stage, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->policy, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->type, _OVER)
SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
SDB_SET_INT64(pRaw, dataPos, pTrans->dbUid, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_DB_FNAME_LEN, _OVER)
SDB_SET_INT32(pRaw, dataPos, redoLogNum, _OVER)
SDB_SET_INT32(pRaw, dataPos, undoLogNum, _OVER)
SDB_SET_INT32(pRaw, dataPos, commitLogNum, _OVER)
SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
for (int32_t i = 0; i < redoLogNum; ++i) {
SSdbRaw *pTmp = taosArrayGetP(pTrans->redoLogs, i);
int32_t len = sdbGetRawTotalSize(pTmp);
SDB_SET_INT32(pRaw, dataPos, len, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, _OVER)
}
for (int32_t i = 0; i < undoLogNum; ++i) {
SSdbRaw *pTmp = taosArrayGetP(pTrans->undoLogs, i);
int32_t len = sdbGetRawTotalSize(pTmp);
SDB_SET_INT32(pRaw, dataPos, len, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, _OVER)
}
for (int32_t i = 0; i < commitLogNum; ++i) {
SSdbRaw *pTmp = taosArrayGetP(pTrans->commitLogs, i);
int32_t len = sdbGetRawTotalSize(pTmp);
SDB_SET_INT32(pRaw, dataPos, len, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pTmp, len, _OVER)
}
for (int32_t i = 0; i < redoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
}
for (int32_t i = 0; i < undoActionNum; ++i) {
STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), TRANS_ENCODE_OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pCont, pAction->contLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pCont, pAction->contLen, _OVER)
}
SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, TRANS_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
if (pTrans->param != NULL) {
SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, TRANS_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
}
SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, TRANS_ENCODE_OVER)
SDB_SET_DATALEN(pRaw, dataPos, TRANS_ENCODE_OVER)
SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
SDB_SET_DATALEN(pRaw, dataPos, _OVER)
terrno = 0;
TRANS_ENCODE_OVER:
_OVER:
if (terrno != 0) {
mError("trans:%d, failed to encode to raw:%p len:%d since %s", pTrans->id, pRaw, dataPos, terrstr());
sdbFreeRaw(pRaw);
@ -229,15 +242,15 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
int16_t type = 0;
int16_t policy = 0;
int16_t stage = 0;
SDB_GET_INT16(pRaw, dataPos, &policy, _OVER)
int16_t policy = 0;
int16_t type = 0;
SDB_GET_INT16(pRaw, dataPos, &stage, _OVER)
SDB_GET_INT16(pRaw, dataPos, &policy, _OVER)
SDB_GET_INT16(pRaw, dataPos, &type, _OVER)
pTrans->policy = policy;
pTrans->stage = stage;
pTrans->transType = type;
pTrans->policy = policy;
pTrans->type = type;
SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
SDB_GET_INT64(pRaw, dataPos, &pTrans->dbUid, _OVER)
SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_DB_FNAME_LEN, _OVER)
@ -400,6 +413,16 @@ static const char *mndTransType(ETrnType type) {
return "subscribe";
case TRN_TYPE_REBALANCE:
return "rebalance";
case TRN_TYPE_COMMIT_OFFSET:
return "commit-offset";
case TRN_TYPE_CREATE_STREAM:
return "create-stream";
case TRN_TYPE_DROP_STREAM:
return "drop-stream";
case TRN_TYPE_CONSUMER_LOST:
return "consumer-lost";
case TRN_TYPE_CONSUMER_RECOVER:
return "consumer-recover";
case TRN_TYPE_CREATE_DNODE:
return "create-qnode";
case TRN_TYPE_DROP_DNODE:
@ -453,7 +476,6 @@ static TransCbFp mndTransGetCbFp(ETrnFuncType ftype) {
}
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
// pTrans->stage = TRN_STAGE_PREPARE;
mTrace("trans:%d, perform insert action, row:%p stage:%s", pTrans->id, pTrans, mndTransStr(pTrans->stage));
if (pTrans->startFunc > 0) {
@ -516,8 +538,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
}
static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
SSdb *pSdb = pMnode->pSdb;
STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId);
STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
if (pTrans == NULL) {
terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
}
@ -540,7 +561,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
pTrans->id = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
pTrans->stage = TRN_STAGE_PREPARE;
pTrans->policy = policy;
pTrans->transType = type;
pTrans->type = type;
pTrans->createdTime = taosGetTimestampMs();
pTrans->rpcHandle = pReq->handle;
pTrans->rpcAHandle = pReq->ahandle;
@ -558,7 +579,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
return NULL;
}
mDebug("trans:%d, local var is created, data:%p", pTrans->id, pTrans);
mDebug("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
return pTrans;
}
@ -585,14 +606,14 @@ static void mndTransDropActions(SArray *pArray) {
void mndTransDrop(STrans *pTrans) {
if (pTrans != NULL) {
mndTransDropData(pTrans);
mDebug("trans:%d, local var is freed, data:%p", pTrans->id, pTrans);
mDebug("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
taosMemoryFreeClear(pTrans);
}
}
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw) {
if (pArray == NULL || pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
terrno = TSDB_CODE_INVALID_PARA;
return -1;
}
@ -674,27 +695,27 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
}
static bool mndIsBasicTrans(STrans *pTrans) {
return pTrans->stage > TRN_TYPE_BASIC_SCOPE && pTrans->stage < TRN_TYPE_BASIC_SCOPE_END;
return pTrans->type > TRN_TYPE_BASIC_SCOPE && pTrans->type < TRN_TYPE_BASIC_SCOPE_END;
}
static bool mndIsGlobalTrans(STrans *pTrans) {
return pTrans->stage > TRN_TYPE_GLOBAL_SCOPE && pTrans->stage < TRN_TYPE_GLOBAL_SCOPE_END;
return pTrans->type > TRN_TYPE_GLOBAL_SCOPE && pTrans->type < TRN_TYPE_GLOBAL_SCOPE_END;
}
static bool mndIsDbTrans(STrans *pTrans) {
return pTrans->stage > TRN_TYPE_DB_SCOPE && pTrans->stage < TRN_TYPE_DB_SCOPE_END;
return pTrans->type > TRN_TYPE_DB_SCOPE && pTrans->type < TRN_TYPE_DB_SCOPE_END;
}
static bool mndIsStbTrans(STrans *pTrans) {
return pTrans->stage > TRN_TYPE_STB_SCOPE && pTrans->stage < TRN_TYPE_STB_SCOPE_END;
return pTrans->type > TRN_TYPE_STB_SCOPE && pTrans->type < TRN_TYPE_STB_SCOPE_END;
}
static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewTrans) {
if (mndIsBasicTrans(pNewTrans)) return 0;
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNewTrans) {
STrans *pTrans = NULL;
void *pIter = NULL;
int32_t code = 0;
bool conflict = false;
if (mndIsBasicTrans(pNewTrans)) return conflict;
while (1) {
pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
@ -703,42 +724,35 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
if (mndIsGlobalTrans(pNewTrans)) {
if (mndIsDbTrans(pTrans) || mndIsStbTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
code = -1;
break;
conflict = true;
} else {
}
}
if (mndIsDbTrans(pNewTrans)) {
if (mndIsBasicTrans(pTrans)) continue;
else if (mndIsDbTrans(pNewTrans)) {
if (mndIsGlobalTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress", pNewTrans->id, pTrans->id);
code = -1;
break;
}
if (mndIsDbTrans(pTrans) || mndIsStbTrans(pTrans)) {
conflict = true;
} else if (mndIsDbTrans(pTrans) || mndIsStbTrans(pTrans)) {
if (pNewTrans->dbUid == pTrans->dbUid) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
code = -1;
break;
conflict = true;
}
} else {
}
}
if (mndIsStbTrans(pNewTrans)) {
if (mndIsBasicTrans(pTrans)) continue;
else if (mndIsStbTrans(pNewTrans)) {
if (mndIsGlobalTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress", pNewTrans->id, pTrans->id);
code = -1;
break;
}
if (mndIsDbTrans(pTrans)) {
conflict = true;
} else if (mndIsDbTrans(pTrans)) {
if (pNewTrans->dbUid == pTrans->dbUid) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
code = -1;
break;
conflict = true;
}
} else {
}
if (mndIsStbTrans(pTrans)) continue;
}
sdbRelease(pMnode->pSdb, pTrans);
@ -746,12 +760,12 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
sdbCancelFetch(pMnode->pSdb, pIter);
sdbRelease(pMnode->pSdb, pTrans);
return code;
return conflict;
}
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
if (mndCheckTransCanBeStartedInParallel(pMnode, pTrans) != 0) {
terrno = TSDB_CODE_MND_TRANS_CANT_PARALLEL;
if (mndCheckTransConflict(pMnode, pTrans)) {
terrno = TSDB_CODE_MND_TRANS_CONFLICT;
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
return -1;
}
@ -805,7 +819,8 @@ static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
}
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
bool sendRsp = false;
bool sendRsp = false;
int32_t code = pTrans->code;
if (pTrans->stage == TRN_STAGE_FINISHED) {
sendRsp = true;
@ -814,12 +829,12 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
if (pTrans->policy == TRN_POLICY_ROLLBACK) {
if (pTrans->stage == TRN_STAGE_UNDO_LOG || pTrans->stage == TRN_STAGE_UNDO_ACTION ||
pTrans->stage == TRN_STAGE_ROLLBACK) {
if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
sendRsp = true;
}
}
if (pTrans->policy == TRN_POLICY_RETRY) {
} else {
if (pTrans->stage == TRN_STAGE_REDO_ACTION && pTrans->failedTimes > 0) {
if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
sendRsp = true;
}
}
@ -831,14 +846,16 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
}
taosMemoryFree(pTrans->rpcRsp);
mDebug("trans:%d, send rsp, code:0x%04x stage:%d app:%p", pTrans->id, pTrans->code & 0xFFFF, pTrans->stage,
mDebug("trans:%d, send rsp, code:0x%04x stage:%d app:%p", pTrans->id, code & 0xFFFF, pTrans->stage,
pTrans->rpcAHandle);
SRpcMsg rspMsg = {.handle = pTrans->rpcHandle,
.code = pTrans->code,
.ahandle = pTrans->rpcAHandle,
.refId = pTrans->rpcRefId,
.pCont = rpcCont,
.contLen = pTrans->rpcRspLen};
SRpcMsg rspMsg = {
.handle = pTrans->rpcHandle,
.ahandle = pTrans->rpcAHandle,
.refId = pTrans->rpcRefId,
.code = code,
.pCont = rpcCont,
.contLen = pTrans->rpcRspLen,
};
tmsgSendRsp(&rspMsg);
pTrans->rpcHandle = NULL;
pTrans->rpcRsp = NULL;
@ -1083,8 +1100,8 @@ static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans) {
} else {
pTrans->code = terrno;
if (pTrans->policy == TRN_POLICY_ROLLBACK) {
pTrans->stage = TRN_STAGE_REDO_ACTION;
mError("trans:%d, stage from commit to redoAction since %s, failedTimes:%d", pTrans->id, terrstr(),
pTrans->stage = TRN_STAGE_UNDO_ACTION;
mError("trans:%d, stage from commit to undoAction since %s, failedTimes:%d", pTrans->id, terrstr(),
pTrans->failedTimes);
continueExec = true;
} else {
@ -1109,7 +1126,7 @@ static bool mndTransPerformCommitLogStage(SMnode *pMnode, STrans *pTrans) {
} else {
pTrans->code = terrno;
pTrans->failedTimes++;
mError("trans:%d, stage keep on commitLog since %s", pTrans->id, terrstr());
mError("trans:%d, stage keep on commitLog since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
continueExec = false;
}
@ -1145,7 +1162,7 @@ static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) {
continueExec = false;
} else {
pTrans->failedTimes++;
mError("trans:%d, stage keep on undoAction since %s", pTrans->id, terrstr());
mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
continueExec = false;
}
@ -1162,7 +1179,7 @@ static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) {
continueExec = true;
} else {
pTrans->failedTimes++;
mError("trans:%d, stage keep on rollback since %s", pTrans->id, terrstr());
mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
continueExec = false;
}
@ -1360,10 +1377,10 @@ static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)dbname, false);
char transType[TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndTransType(pTrans->transType), pShow->pMeta->pSchemas[cols].bytes);
char type[TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndTransType(pTrans->type), pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)transType, false);
colDataAppend(pColInfo, numOfRows, (const char *)type, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false);

3
source/dnode/mnode/impl/src/mndUser.c
View File

@ -275,9 +275,6 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate
}
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
char *param = strdup("====> test code to be deleted later <=====");
mndTransSetCb(pTrans, TEST_TRANS_START_FUNC, TEST_TRANS_STOP_FUNC, param, strlen(param) + 1);
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);

11
source/dnode/mnode/impl/src/mnode.c
View File

@ -43,9 +43,6 @@
#include "mndUser.h"
#include "mndVgroup.h"
#define MQ_TIMER_MS 2000
#define TRNAS_TIMER_MS 6000
static void *mndBuildTimerMsg(int32_t *pContLen) {
SMTimerReq timerReq = {0};
@ -68,7 +65,7 @@ static void mndPullupTrans(void *param, void *tmrId) {
tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
}
taosTmrReset(mndPullupTrans, TRNAS_TIMER_MS, pMnode, pMnode->timer, &pMnode->transTimer);
taosTmrReset(mndPullupTrans, tsTransPullupMs, pMnode, pMnode->timer, &pMnode->transTimer);
}
static void mndCalMqRebalance(void *param, void *tmrId) {
@ -84,7 +81,7 @@ static void mndCalMqRebalance(void *param, void *tmrId) {
tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
}
taosTmrReset(mndCalMqRebalance, MQ_TIMER_MS, pMnode, pMnode->timer, &pMnode->mqTimer);
taosTmrReset(mndCalMqRebalance, tsMaRebalanceMs, pMnode, pMnode->timer, &pMnode->mqTimer);
}
static void mndPullupTelem(void *param, void *tmrId) {
@ -106,12 +103,12 @@ static int32_t mndInitTimer(SMnode *pMnode) {
return -1;
}
if (taosTmrReset(mndPullupTrans, TRNAS_TIMER_MS, pMnode, pMnode->timer, &pMnode->transTimer)) {
if (taosTmrReset(mndPullupTrans, tsTransPullupMs, pMnode, pMnode->timer, &pMnode->transTimer)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (taosTmrReset(mndCalMqRebalance, MQ_TIMER_MS, pMnode, pMnode->timer, &pMnode->mqTimer)) {
if (taosTmrReset(mndCalMqRebalance, tsMaRebalanceMs, pMnode, pMnode->timer, &pMnode->mqTimer)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}

47
source/dnode/mnode/impl/test/trans/trans2.cpp
View File

@ -19,10 +19,10 @@ void reportStartup(SMgmtWrapper *pWrapper, const char *name, const char *desc) {
class MndTestTrans2 : public ::testing::Test {
protected:
static void SetUpTestSuite() {
static void InitLog() {
dDebugFlag = 143;
vDebugFlag = 0;
mDebugFlag = 143;
mDebugFlag = 207;
cDebugFlag = 0;
jniDebugFlag = 0;
tmrDebugFlag = 135;
@ -42,9 +42,9 @@ class MndTestTrans2 : public ::testing::Test {
if (taosInitLog("taosdlog", 1) != 0) {
printf("failed to init log file\n");
}
}
walInit();
static void InitMnode() {
static SMsgCb msgCb = {0};
msgCb.reportStartupFp = reportStartup;
msgCb.pWrapper = (SMgmtWrapper *)(&msgCb); // hack
@ -58,12 +58,22 @@ class MndTestTrans2 : public ::testing::Test {
strcpy(opt.replicas[0].fqdn, "localhost");
opt.msgCb = msgCb;
tsTransPullupMs = 1000;
const char *mnodepath = "/tmp/mnode_test_trans";
taosRemoveDir(mnodepath);
pMnode = mndOpen(mnodepath, &opt);
mndStart(pMnode);
}
static void SetUpTestSuite() {
InitLog();
walInit();
InitMnode();
}
static void TearDownTestSuite() {
mndStop(pMnode);
mndClose(pMnode);
walCleanUp();
taosCloseLog();
@ -76,11 +86,11 @@ class MndTestTrans2 : public ::testing::Test {
void SetUp() override {}
void TearDown() override {}
void CreateUser(const char *user) {
int32_t CreateUser(const char *acct, const char *user) {
SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
tstrncpy(userObj.user, user, TSDB_USER_LEN);
tstrncpy(userObj.acct, "root", TSDB_USER_LEN);
tstrncpy(userObj.acct, acct, TSDB_USER_LEN);
userObj.createdTime = taosGetTimestampMs();
userObj.updateTime = userObj.createdTime;
userObj.superUser = 1;
@ -94,19 +104,34 @@ class MndTestTrans2 : public ::testing::Test {
char *param = strdup("====> test param <=====");
mndTransSetCb(pTrans, TEST_TRANS_START_FUNC, TEST_TRANS_STOP_FUNC, param, strlen(param) + 1);
mndTransPrepare(pMnode, pTrans);
int32_t code = mndTransPrepare(pMnode, pTrans);
mndTransDrop(pTrans);
return code;
}
};
SMnode *MndTestTrans2::pMnode;
TEST_F(MndTestTrans2, 01_CbFunc) {
const char *acct = "root";
const char *acct_invalid = "root1";
const char *user1 = "test1";
const char *user2 = "test2";
SUserObj *pUser1 = NULL;
SUserObj *pUser2 = NULL;
ASSERT_NE(pMnode, nullptr);
const char *user1 = "test1";
CreateUser(user1);
SUserObj *pUser1 = mndAcquireUser(pMnode, user1);
// create user success
EXPECT_EQ(CreateUser(acct, user1), 0);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser1, nullptr);
// failed to create user and rollback
EXPECT_EQ(CreateUser(acct_invalid, user2), 0);
pUser2 = mndAcquireUser(pMnode, user2);
ASSERT_EQ(pUser2, nullptr);
mndTransPullup(pMnode);
}

37
source/libs/executor/inc/executorimpl.h
View File

@ -119,6 +119,17 @@ typedef struct SLimit {
int64_t offset;
} SLimit;
typedef struct SFileBlockLoadRecorder {
uint64_t totalRows;
uint64_t totalCheckedRows;
uint32_t totalBlocks;
uint32_t loadBlocks;
uint32_t loadBlockStatis;
uint32_t skipBlocks;
uint32_t filterOutBlocks;
uint64_t elapsedTime;
} SFileBlockLoadRecorder;
typedef struct STaskCostInfo {
int64_t created;
int64_t start;
@ -132,14 +143,10 @@ typedef struct STaskCostInfo {
uint64_t loadDataInCacheSize;
uint64_t loadDataTime;
uint64_t totalRows;
uint64_t totalCheckedRows;
uint32_t totalBlocks;
uint32_t loadBlocks;
uint32_t loadBlockStatis;
uint32_t skipBlocks;
uint32_t filterOutBlocks;
SFileBlockLoadRecorder* pRecoder;
uint64_t elapsedTime;
uint64_t firstStageMergeTime;
uint64_t winInfoSize;
uint64_t tableInfoSize;
@ -268,7 +275,7 @@ typedef struct SOperatorFpSet {
typedef struct SOperatorInfo {
uint8_t operatorType;
bool blockingOptr; // block operator or not
bool blocking; // block operator or not
uint8_t status; // denote if current operator is completed
int32_t numOfOutput; // number of columns of the current operator results
char* name; // name, used to show the query execution plan
@ -333,17 +340,14 @@ typedef struct SScanInfo {
typedef struct STableScanInfo {
void* dataReader;
int32_t numOfBlocks; // extract basic running information.
int32_t numOfSkipped;
int32_t numOfBlockStatis;
SFileBlockLoadRecorder readRecorder;
int64_t numOfRows;
int64_t elapsedTime;
int32_t prevGroupId; // previous table group id
// int32_t prevGroupId; // previous table group id
SScanInfo scanInfo;
int32_t current;
SNode* pFilterNode; // filter operator info
SqlFunctionCtx* pCtx; // next operator query context
int32_t scanTimes;
SNode* pFilterNode; // filter info, which is push down by optimizer
SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context
SResultRowInfo* pResultRowInfo;
int32_t* rowCellInfoOffset;
SExprInfo* pExpr;
@ -397,7 +401,6 @@ typedef struct SSysTableScanInfo {
SArray* scanCols; // SArray<int16_t> scan column id list
SName name;
SSDataBlock* pRes;
int32_t capacity;
int64_t numOfBlocks; // extract basic running information.
SLoadRemoteDataInfo loadInfo;
} SSysTableScanInfo;

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

@ -132,6 +132,7 @@ void doSetOperatorCompleted(SOperatorInfo* pOperator) {
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator) {
OPTR_SET_OPENED(pOperator);
pOperator->cost.openCost = 0;
return TSDB_CODE_SUCCESS;
}
@ -1592,8 +1593,8 @@ int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableSc
STaskCostInfo* pCost = &pTaskInfo->cost;
pCost->totalBlocks += 1;
pCost->totalRows += pBlock->info.rows;
// pCost->totalBlocks += 1;
// pCost->totalRows += pBlock->info.rows;
#if 0
// Calculate all time windows that are overlapping or contain current data block.
// If current data block is contained by all possible time window, do not load current data block.
@ -2411,12 +2412,13 @@ void queryCostStatis(SExecTaskInfo* pTaskInfo) {
//
// calculateOperatorProfResults(pQInfo);
qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64
" us, total blocks:%d, "
"load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pSummary->totalBlocks,
pSummary->loadBlockStatis, pSummary->loadBlocks, pSummary->totalRows, pSummary->totalCheckedRows);
//
SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
if (pSummary->pRecoder != NULL) {
qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64 " us, total blocks:%d, "
"load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pRecorder->totalBlocks,
pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows, pRecorder->totalCheckedRows);
}
// qDebug("QInfo:0x%"PRIx64" :cost summary: winResPool size:%.2f Kb, numOfWin:%"PRId64", tableInfoSize:%.2f Kb,
// hashTable:%.2f Kb", pQInfo->qId, pSummary->winInfoSize/1024.0,
// pSummary->numOfTimeWindows, pSummary->tableInfoSize/1024.0, pSummary->hashSize/1024.0);
@ -3282,7 +3284,7 @@ SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock
pOperator->name = "ExchangeOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;
pOperator->blockingOptr = false;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = pBlock->info.numOfCols;
@ -3673,7 +3675,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
pOperator->name = "SortedMerge";
// pOperator->operatorType = OP_SortedMerge;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = num;
@ -3756,7 +3758,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR
pOperator->name = "SortOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
@ -4419,7 +4421,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pOperator->name = "TableAggregate";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_AGG;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->pExpr = pExprInfo;
@ -4532,7 +4534,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* p
pOperator->name = "ProjectOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT;
pOperator->blockingOptr = false;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->pExpr = pExprInfo;
@ -4615,7 +4617,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExp
}
pOperator->name = "FillOperator";
pOperator->blockingOptr = false;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL;
pOperator->pExpr = pExpr;
@ -4861,15 +4863,19 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
SSDataBlock* pResBlock = createResDataBlock(pScanPhyNode->node.pOutputDataBlockDesc);
SQueryTableDataCond cond = {0};
int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
if (code != TSDB_CODE_SUCCESS) {
return NULL;
}
SInterval interval = extractIntervalInfo(pTableScanNode);
return createTableScanOperatorInfo(pDataReader, &cond, numOfCols, pTableScanNode->dataRequired,
SOperatorInfo* pOperator = createTableScanOperatorInfo(pDataReader, &cond, numOfCols, pTableScanNode->dataRequired,
pTableScanNode->scanSeq, pColList, pResBlock, pScanPhyNode->node.pConditions,
&interval, pTableScanNode->ratio, pTaskInfo);
STableScanInfo* pScanInfo = pOperator->info;
pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
return pOperator;
} else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pPhyNode;
SSDataBlock* pResBlock = createResDataBlock(pExchange->node.pOutputDataBlockDesc);
@ -4877,8 +4883,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
} else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode; // simple child table.
int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo,
queryId, taskId);
int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo, queryId, taskId);
SArray* tableIdList = extractTableIdList(pTableGroupInfo);
SSDataBlock* pResBlock = createResDataBlock(pScanPhyNode->node.pOutputDataBlockDesc);
@ -5628,7 +5633,7 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOf
pOperator->name = "JoinOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_JOIN;
pOperator->blockingOptr = false;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols;

4
source/libs/executor/src/groupoperator.c
View File

@ -353,7 +353,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
pOperator->name = "GroupbyAggOperator";
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
// pOperator->operatorType = OP_Groupby;
pOperator->pExpr = pExprInfo;
@ -612,7 +612,7 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo*
}
pOperator->name = "PartitionOperator";
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION;
pInfo->binfo.pRes = pResultBlock;

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

@ -55,13 +55,32 @@ typedef struct SIFParam {
SArray *result;
char * condValue;
uint8_t colValType;
col_id_t colId;
int64_t suid; // add later
char dbName[TSDB_DB_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
int64_t suid; // add later
char dbName[TSDB_DB_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
} SIFParam;
static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
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) {
*dst = QUERY_REGEX;
} else {
return TSDB_CODE_QRY_INVALID_INPUT;
}
return TSDB_CODE_SUCCESS;
}
typedef int32_t (*sif_func_t)(SIFParam *left, SIFParam *rigth, SIFParam *output);
// construct tag filter operator later
static void destroyTagFilterOperatorInfo(void *param) {
@ -145,10 +164,11 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
SColumnNode *cn = (SColumnNode *)node;
/*only support tag column*/
SIF_ERR_RET(sifValidateColumn(cn));
param->colId = cn->colId;
param->colValType = cn->node.resType.type;
memcpy(param->dbName, cn->dbName, sizeof(cn->dbName));
memcpy(param->colName, cn->colName, sizeof(cn->colName));
break;
}
case QUERY_NODE_NODE_LIST: {
@ -231,61 +251,73 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu
qError("index-filter not support buildin function");
return TSDB_CODE_QRY_INVALID_INPUT;
}
static int32_t sifIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) {
SIndexMultiTermQuery *mq = indexMultiTermQueryCreate(MUST);
return TSDB_CODE_SUCCESS;
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 (tm == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST);
EIndexQueryType qtype = 0;
SIF_ERR_RET(sifGetFuncFromSql(operType, &qtype));
indexMultiTermQueryAdd(mtm, tm, qtype);
int ret = indexSearch(NULL, mtm, output->result);
indexMultiTermQueryDestroy(mtm);
return ret;
}
static int32_t sifLessThanFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_LOWER_THAN;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifLessEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_LOWER_EQUAL;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifGreaterThanFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_GREATER_THAN;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifGreaterEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_GREATER_EQUAL;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_EQUAL;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifNotEqualFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_NOT_EQUAL;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifInFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_IN;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifNotInFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_NOT_IN;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifLikeFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_LIKE;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifNotLikeFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_NOT_LIKE;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_MATCH;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifNotMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
int id = OP_TYPE_NMATCH;
return sifIndex(left, right, id, output);
return sifDoIndex(left, right, id, output);
}
static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
// add more except
@ -460,6 +492,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
qError("index-filter failed to taosHashInit");
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
nodesWalkExprPostOrder(pNode, sifCalcWalker, &ctx);
SIF_ERR_RET(ctx.code);
@ -498,6 +531,7 @@ SIdxFltStatus idxGetFltStatus(SNode *pFilterNode) {
if (pFilterNode == NULL) {
return SFLT_NOT_INDEX;
}
// impl later
return SFLT_ACCURATE_INDEX;
}

267
source/libs/executor/src/scanoperator.c
View File

@ -13,7 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "common/ttime.h"
#include "ttime.h"
#include "filter.h"
#include "function.h"
#include "functionMgt.h"
@ -36,10 +36,10 @@
#define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
#define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo);
static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity);
static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size, const char* dbName);
int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size, const char* dbName);
void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
for (int32_t i = 0; i < numOfOutput; ++i) {
SWITCH_ORDER(pCtx[i].order);
}
@ -158,11 +158,11 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
return false;
}
int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableScanInfo* pInfo = pOperator->info;
STaskCostInfo* pCost = &pTaskInfo->cost;
SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;
pCost->totalBlocks += 1;
pCost->totalRows += pBlock->info.rows;
@ -188,11 +188,11 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
} else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
pCost->loadBlockStatis += 1;
bool allHave = true;
bool allColumnsHaveAgg = true;
SColumnDataAgg** pColAgg = NULL;
tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allHave);
tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allColumnsHaveAgg);
if (allHave == true) {
if (allColumnsHaveAgg == true) {
int32_t numOfCols = pBlock->info.numOfCols;
// todo create this buffer during creating operator
@ -236,6 +236,7 @@ int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo,
}
relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols);
// todo record the filter time cost
doFilter(pTableScanInfo->pFilterNode, pBlock);
if (pBlock->info.rows == 0) {
pCost->filterOutBlocks += 1;
@ -266,7 +267,6 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
pTableScanInfo->numOfBlocks += 1;
tsdbRetrieveDataBlockInfo(pTableScanInfo->dataReader, &pBlock->info);
uint32_t status = 0;
@ -296,15 +296,16 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
return NULL;
}
while (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) {
// do the ascending order traverse in the first place.
while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
SSDataBlock* p = doTableScanImpl(pOperator);
if (p != NULL) {
return p;
}
pTableScanInfo->current += 1;
pTableScanInfo->scanTimes += 1;
if (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) {
if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN;
@ -319,7 +320,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
}
int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
if (pTableScanInfo->current < total) {
if (pTableScanInfo->scanTimes < total) {
if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, pTableScanInfo->numOfOutput);
tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
@ -329,21 +330,20 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
qDebug("%s start to descending order scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64,
GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
while (pTableScanInfo->current < total) {
while (pTableScanInfo->scanTimes < total) {
SSDataBlock* p = doTableScanImpl(pOperator);
if (p != NULL) {
return p;
}
pTableScanInfo->current += 1;
pTableScanInfo->scanTimes += 1;
if (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) {
if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN;
qDebug("%s start to repeat descending order scan data blocks due to query func required, qrange:%" PRId64
"-%" PRId64,
GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
"-%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey);
// do prepare for the next round table scan operation
tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
@ -372,48 +372,50 @@ SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCon
pInfo->cond = *pCond;
pInfo->scanInfo = (SScanInfo){.numOfAsc = scanInfo[0], .numOfDesc = scanInfo[1]};
pInfo->interval = *pInterval;
pInfo->sampleRatio = sampleRatio;
pInfo->interval = *pInterval;
pInfo->sampleRatio = sampleRatio;
pInfo->dataBlockLoadFlag = dataLoadFlag;
pInfo->pResBlock = pResBlock;
pInfo->pFilterNode = pCondition;
pInfo->dataReader = pDataReader;
pInfo->current = 0;
pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColMatchInfo;
pInfo->pResBlock = pResBlock;
pInfo->pFilterNode = pCondition;
pInfo->dataReader = pDataReader;
pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColMatchInfo;
pOperator->name = "TableScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = numOfOutput;
pOperator->fpSet.getNextFn = doTableScan;
pOperator->pTaskInfo = pTaskInfo;
pOperator->name = "TableScanOperator"; // for dubug purpose
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = numOfOutput;
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, NULL, NULL, NULL, NULL);
static int32_t cost = 0;
pOperator->cost.openCost = ++cost;
// for non-blocking operator, the open cost is always 0
pOperator->cost.openCost = 0;
pOperator->cost.totalCost = ++cost;
pOperator->resultInfo.totalRows = ++cost;
return pOperator;
}
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle) {
SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {
STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
pInfo->dataReader = pTsdbReadHandle;
pInfo->current = 0;
pInfo->prevGroupId = -1;
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
pOperator->name = "TableSeqScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN;
pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->fpSet.getNextFn = doTableScanImpl;
pInfo->dataReader = pReadHandle;
// pInfo->prevGroupId = -1;
pOperator->name = "TableSeqScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL, NULL, NULL, NULL);
return pOperator;
}
@ -482,7 +484,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo*
pOperator->name = "DataBlockInfoScanOperator";
// pOperator->operatorType = OP_TableBlockInfoScan;
pOperator->blockingOptr = false;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->fpSet._openFn = operatorDummyOpenFn;
pOperator->fpSet.getNextFn = doBlockInfoScan;
@ -636,20 +638,22 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock*
}
pInfo->readerHandle = streamReadHandle;
pInfo->pRes = pResBlock;
pInfo->pCondition = pCondition;
pInfo->pRes = pResBlock;
pInfo->pCondition = pCondition;
pOperator->name = "StreamBlockScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->fpSet._openFn = operatorDummyOpenFn;
pOperator->name = "StreamBlockScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->fpSet._openFn = operatorDummyOpenFn;
pOperator->fpSet.getNextFn = doStreamBlockScan;
pOperator->fpSet.closeFn = operatorDummyCloseFn;
pOperator->fpSet.closeFn = operatorDummyCloseFn;
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamBlockScan, NULL, NULL, operatorDummyCloseFn, NULL, NULL, NULL);
return pOperator;
}
@ -671,14 +675,12 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
switch (nType) {
case QUERY_NODE_OPERATOR: {
SOperatorNode* node = (SOperatorNode*)pNode;
if (OP_TYPE_EQUAL == node->opType) {
*(int32_t*)pContext = 1;
return DEAL_RES_CONTINUE;
}
*(int32_t*)pContext = 0;
return DEAL_RES_IGNORE_CHILD;
}
case QUERY_NODE_COLUMN: {
@ -709,19 +711,17 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) {
default:
break;
}
return DEAL_RES_CONTINUE;
}
void getDBNameFromCondition(SNode* pCondition, char* dbName) {
static void getDBNameFromCondition(SNode* pCondition, const char* dbName) {
if (NULL == pCondition) {
return;
}
nodesWalkExpr(pCondition, getDBNameFromConditionWalker, dbName);
nodesWalkExpr(pCondition, getDBNameFromConditionWalker, (char*) dbName);
}
static int32_t loadSysTableContentCb(void* param, const SDataBuf* pMsg, int32_t code) {
static int32_t loadSysTableCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SOperatorInfo* operator=(SOperatorInfo*) param;
SSysTableScanInfo* pScanResInfo = (SSysTableScanInfo*)operator->info;
if (TSDB_CODE_SUCCESS == code) {
@ -746,13 +746,14 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
}
SFilterInfo* filter = NULL;
int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0);
int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0);
SFilterColumnParam param1 = {.numOfCols = pInfo->pRes->info.numOfCols, .pDataBlock = pInfo->pRes->pDataBlock};
code = filterSetDataFromSlotId(filter, &param1);
int8_t* rowRes = NULL;
bool keep = filterExecute(filter, pInfo->pRes, &rowRes, NULL, param1.numOfCols);
bool keep = filterExecute(filter, pInfo->pRes, &rowRes, NULL, param1.numOfCols);
filterFreeInfo(filter);
SSDataBlock* px = createOneDataBlock(pInfo->pRes, false);
@ -795,61 +796,31 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
static SSDataBlock* buildSysTableMetaBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = 10;
pBlock->info.hasVarCol = true;
size_t size = 0;
const SSysTableMeta *pMeta = NULL;
getInfosDbMeta(&pMeta, &size);
int32_t index = 0;
for(int32_t i = 0; i < size; ++i) {
if(strcmp(pMeta[i].name, TSDB_INS_TABLE_USER_TABLES) == 0) {
index = i;
break;
}
}
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
SColumnInfoData colInfoData = {0};
colInfoData.info.colId = 1;
colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR;
colInfoData.info.bytes = (TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
for(int32_t i = 0; i < pMeta[index].colNum; ++i) {
SColumnInfoData colInfoData = {0};
colInfoData.info.colId = i + 1;
colInfoData.info.type = pMeta[index].schema[i].type;
colInfoData.info.bytes = pMeta[index].schema[i].bytes;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
}
colInfoData.info.colId = 2;
colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR;
colInfoData.info.bytes = (TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 3;
colInfoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
colInfoData.info.bytes = 8;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 4;
colInfoData.info.type = TSDB_DATA_TYPE_INT;
colInfoData.info.bytes = 4;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 5;
colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR;
colInfoData.info.bytes = (TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 6;
colInfoData.info.type = TSDB_DATA_TYPE_BIGINT;
colInfoData.info.bytes = 8;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 7;
colInfoData.info.type = TSDB_DATA_TYPE_INT;
colInfoData.info.bytes = 4;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 8;
colInfoData.info.type = TSDB_DATA_TYPE_INT;
colInfoData.info.bytes = 4;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 9;
colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR;
colInfoData.info.bytes = 512 + VARSTR_HEADER_SIZE;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
colInfoData.info.colId = 10;
colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR;
colInfoData.info.bytes = 20 + VARSTR_HEADER_SIZE;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
pBlock->info.numOfCols = pMeta[index].colNum;
pBlock->info.hasVarCol = true;
return pBlock;
}
@ -868,7 +839,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
return NULL;
}
buildSysDbTableInfo(pInfo);
buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity);
doFilterResult(pInfo);
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
@ -896,7 +867,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
varDataSetLen(dbname, strlen(varDataVal(dbname)));
SSDataBlock* p = buildSysTableMetaBlock();
blockDataEnsureCapacity(p, pInfo->capacity);
blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);
char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
while (metaTbCursorNext(pInfo->pCur) == 0) {
@ -977,7 +948,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
pColInfoData = taosArrayGet(p->pDataBlock, 9);
colDataAppend(pColInfoData, numOfRows, str, false);
if (++numOfRows >= pInfo->capacity) {
if (++numOfRows >= pOperator->resultInfo.capacity) {
break;
}
}
@ -1022,7 +993,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
pMsgSendInfo->msgInfo.pData = buf1;
pMsgSendInfo->msgInfo.len = contLen;
pMsgSendInfo->msgType = TDMT_MND_SYSTABLE_RETRIEVE;
pMsgSendInfo->fp = loadSysTableContentCb;
pMsgSendInfo->fp = loadSysTableCallback;
int64_t transporterId = 0;
int32_t code = asyncSendMsgToServer(pInfo->pTransporter, &pInfo->epSet, &transporterId, pMsgSendInfo);
@ -1060,9 +1031,9 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
}
}
int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo) {
int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity) {
SSDataBlock* p = buildSysTableMetaBlock();
blockDataEnsureCapacity(p, pInfo->capacity);
blockDataEnsureCapacity(p, capacity);
size_t size = 0;
const SSysTableMeta* pSysDbTableMeta = NULL;
@ -1133,18 +1104,19 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe
return NULL;
}
pInfo->accountId = accountId;
pInfo->accountId = accountId;
pInfo->showRewrite = showRewrite;
pInfo->pRes = pResBlock;
pInfo->capacity = 4096;
pInfo->pCondition = pCondition;
pInfo->scanCols = colList;
pInfo->pRes = pResBlock;
pInfo->pCondition = pCondition;
pInfo->scanCols = colList;
initResultSizeInfo(pOperator, 4096);
tNameAssign(&pInfo->name, pName);
const char* name = tNameGetTableName(&pInfo->name);
if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
pInfo->readHandle = *(SReadHandle*) readHandle;
blockDataEnsureCapacity(pInfo->pRes, pInfo->capacity);
blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
} else {
tsem_init(&pInfo->ready, 0, 0);
pInfo->epSet = epset;
@ -1173,12 +1145,12 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe
#endif
}
pOperator->name = "SysTableScanOperator";
pOperator->name = "SysTableScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN;
pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->numOfOutput = pResBlock->info.numOfCols;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator,
NULL, NULL, NULL);
pOperator->pTaskInfo = pTaskInfo;
@ -1188,7 +1160,6 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe
static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
#if 0
SOperatorInfo* pOperator = (SOperatorInfo*) param;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
@ -1322,29 +1293,29 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
pInfo->pRes = blockDataDestroy(pInfo->pRes);
}
SOperatorInfo* createTagScanOperatorInfo(void* pReaderHandle, SExprInfo* pExpr, int32_t numOfOutput,
SExecTaskInfo* pTaskInfo) {
SOperatorInfo* createTagScanOperatorInfo(void* readHandle, SExprInfo* pExpr, int32_t numOfOutput, SExecTaskInfo* pTaskInfo) {
STagScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
goto _error;
}
pInfo->pReader = pReaderHandle;
pInfo->curPos = 0;
pOperator->name = "TagScanOperator";
pInfo->pReader = readHandle;
pInfo->curPos = 0;
pOperator->name = "TagScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
pOperator->blockingOptr = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->pExpr = pExpr;
pOperator->numOfOutput = numOfOutput;
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, NULL, destroyTagScanOperatorInfo, NULL, NULL, NULL);
pOperator->pExpr = pExpr;
pOperator->numOfOutput = numOfOutput;
pOperator->pTaskInfo = pTaskInfo;
return pOperator;
_error:
taosMemoryFree(pInfo);
taosMemoryFree(pOperator);

10
source/libs/executor/src/timewindowoperator.c
View File

@ -1078,7 +1078,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pOperator->name = "TimeIntervalAggOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERVAL;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->pTaskInfo = pTaskInfo;
@ -1137,7 +1137,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr
pOperator->name = "StreamTimeIntervalAggOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERVAL;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->pTaskInfo = pTaskInfo;
@ -1343,7 +1343,7 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pOperator->name = "TimeSliceOperator";
// pOperator->operatorType = OP_AllTimeWindow;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols;
@ -1385,7 +1385,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf
pInfo->tsSlotId = tsSlotId;
pOperator->name = "StateWindowOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExpr;
pOperator->numOfOutput = numOfCols;
@ -1437,7 +1437,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo
pInfo->reptScan = false;
pOperator->name = "SessionWindowAggOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW;
pOperator->blockingOptr = true;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols;

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;

5
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 {
@ -57,7 +58,7 @@ struct SIndex {
char* path;
SIndexStat stat;
SIndexStat stat;
TdThreadMutex mtx;
};
@ -165,7 +166,7 @@ int32_t indexSerialCacheKey(ICacheKey* key, char* buf);
} while (0)
#define INDEX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0)
#define INDEX_TYPE_GET_TYPE(ty) (ty & 0x0F)
#define INDEX_TYPE_GET_TYPE(ty) (ty & 0x0F)
#define INDEX_TYPE_ADD_EXTERN_TYPE(ty, exTy) \
do { \
uint8_t oldTy = ty; \

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

@ -175,55 +175,19 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) {
return 0;
}
int indexSearch(SIndex* index, SIndexMultiTermQuery* multiQuerys, SArray* result) {
#ifdef USE_LUCENE
EIndexOperatorType opera = multiQuerys->opera;
int nQuery = taosArrayGetSize(multiQuerys->query);
char** fields = taosMemoryMalloc(sizeof(char*) * nQuery);
char** keys = taosMemoryMalloc(sizeof(char*) * nQuery);
int* types = taosMemoryMalloc(sizeof(int) * nQuery);
for (int i = 0; i < nQuery; i++) {
SIndexTermQuery* p = taosArrayGet(multiQuerys->query, i);
SIndexTerm* term = p->field_value;
fields[i] = taosMemoryCalloc(1, term->nKey + 1);
keys[i] = taosMemoryCalloc(1, term->nVal + 1);
memcpy(fields[i], term->key, term->nKey);
memcpy(keys[i], term->val, term->nVal);
types[i] = (int)(p->type);
}
int* tResult = NULL;
int tsz = 0;
index_multi_search(index->index, (const char**)fields, (const char**)keys, types, nQuery, opera, &tResult, &tsz);
for (int i = 0; i < tsz; i++) {
taosArrayPush(result, &tResult[i]);
}
for (int i = 0; i < nQuery; i++) {
taosMemoryFree(fields[i]);
taosMemoryFree(keys[i]);
}
taosMemoryFree(fields);
taosMemoryFree(keys);
taosMemoryFree(types);
#endif
#ifdef USE_INVERTED_INDEX
EIndexOperatorType opera = multiQuerys->opera; // relation of querys
SArray* interResults = taosArrayInit(4, POINTER_BYTES);
SArray* iRslts = taosArrayInit(4, POINTER_BYTES);
int nQuery = taosArrayGetSize(multiQuerys->query);
for (size_t i = 0; i < nQuery; i++) {
SIndexTermQuery* qTerm = taosArrayGet(multiQuerys->query, i);
SArray* tResult = NULL;
indexTermSearch(index, qTerm, &tResult);
taosArrayPush(interResults, (void*)&tResult);
SIndexTermQuery* qterm = taosArrayGet(multiQuerys->query, i);
SArray* trslt = NULL;
indexTermSearch(index, qterm, &trslt);
taosArrayPush(iRslts, (void*)&trslt);
}
indexMergeFinalResults(interResults, opera, result);
indexInterResultsDestroy(interResults);
indexMergeFinalResults(iRslts, opera, result);
indexInterResultsDestroy(iRslts);
#endif
return 0;
@ -280,8 +244,8 @@ int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EInde
return 0;
}
SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char* colName,
int32_t nColName, const char* colVal, int32_t nColVal) {
SIndexTerm* indexTermCreate(int64_t suid, SIndexOperOnColumn oper, int8_t queryType, uint8_t colType,
const char* colName, int32_t nColName, const char* colVal, int32_t nColVal) {
SIndexTerm* tm = (SIndexTerm*)taosMemoryCalloc(1, (sizeof(SIndexTerm)));
if (tm == NULL) {
return NULL;

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

@ -34,9 +34,129 @@ static char* indexCacheTermGet(const void* pData);
static MemTable* indexInternalCacheCreate(int8_t type);
static int32_t cacheSearchTerm(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* 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, cacheSearchLessThan,
cacheSearchLessEqual, cacheSearchGreaterThan, cacheSearchGreaterEqual, cacheSearchRange};
static void doMergeWork(SSchedMsg* msg);
static bool indexCacheIteratorNext(Iterate* itera);
static int32_t cacheSearchTerm(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
if (cache == NULL) {
return 0;
}
MemTable* mem = cache;
char* key = indexCacheTermGet(ct);
SSkipListIterator* iter = tSkipListCreateIterFromVal(mem->mem, key, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
while (tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
if (node == NULL) {
break;
}
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
if (0 == strcmp(c->colVal, ct->colVal)) {
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 {
break;
}
}
tSkipListDestroyIter(iter);
return 0;
}
static int32_t cacheSearchPrefix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
// impl later
return 0;
}
static int32_t cacheSearchSuffix(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
// impl later
return 0;
}
static int32_t cacheSearchRegex(void* cache, CacheTerm* ct, SIdxTempResult* tr, STermValueType* s) {
// 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;
}
static IterateValue* indexCacheIteratorGetValue(Iterate* iter);
IndexCache* indexCacheCreate(SIndex* idx, uint64_t suid, const char* colName, int8_t type) {
@ -263,33 +383,7 @@ static int indexQueryMem(MemTable* mem, CacheTerm* ct, EIndexQueryType qtype, SI
if (mem == NULL) {
return 0;
}
char* key = indexCacheTermGet(ct);
SSkipListIterator* iter = tSkipListCreateIterFromVal(mem->mem, key, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
while (tSkipListIterNext(iter)) {
SSkipListNode* node = tSkipListIterGet(iter);
if (node != NULL) {
CacheTerm* c = (CacheTerm*)SL_GET_NODE_DATA(node);
if (qtype == QUERY_TERM) {
if (0 == strcmp(c->colVal, ct->colVal)) {
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 {
break;
}
} else if (qtype == QUERY_PREFIX) {
} else if (qtype == QUERY_SUFFIX) {
} else if (qtype == QUERY_RANGE) {
}
}
}
tSkipListDestroyIter(iter);
return 0;
return cacheSearch[qtype](mem, ct, tr, s);
}
int indexCacheSearch(void* cache, SIndexTermQuery* query, SIdxTempResult* result, STermValueType* s) {
int64_t st = taosGetTimestampUs();

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

@ -57,6 +57,24 @@ static int tfileCompare(const void* a, const void* b);
static int tfileParseFileName(const char* filename, uint64_t* suid, char* col, int* version);
static void tfileGenFileName(char* filename, uint64_t suid, const char* col, int version);
static void tfileGenFileFullName(char* fullname, const char* path, uint64_t suid, const char* col, int32_t version);
/*
* search from tfile
*/
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, tfSearchLessThan,
tfSearchLessEqual, tfSearchGreaterThan, tfSearchGreaterEqual, tfSearchRange};
TFileCache* tfileCacheCreate(const char* path) {
TFileCache* tcache = taosMemoryCalloc(1, sizeof(TFileCache));
@ -183,59 +201,194 @@ void tfileReaderDestroy(TFileReader* reader) {
writerCtxDestroy(reader->ctx, reader->remove);
taosMemoryFree(reader);
}
static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
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);
}
int64_t st = taosGetTimestampUs();
FstSlice key = fstSliceCreate(p, sz);
uint64_t offset;
if (fstGet(((TFileReader*)reader)->fst, &key, &offset)) {
int64_t et = taosGetTimestampUs();
int64_t cost = et - st;
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
tem->suid, tem->colName, tem->colVal, cost);
ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
cost = taosGetTimestampUs() - et;
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
tem->colName, tem->colVal, cost);
}
if (hasJson) {
taosMemoryFree(p);
}
fstSliceDestroy(&key);
return 0;
}
static int32_t tfSearchPrefix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(tem->colType, TSDB_DATA_TYPE_JSON);
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_PREFIX);
FstStreamBuilder* sb = fstSearch(((TFileReader*)reader)->fst, ctx);
StreamWithState* st = streamBuilderIntoStream(sb);
StreamWithStateResult* rt = NULL;
while ((rt = streamWithStateNextWith(st, NULL)) != NULL) {
taosArrayPush(offsets, &(rt->out.out));
swsResultDestroy(rt);
}
streamWithStateDestroy(st);
fstStreamBuilderDestroy(sb);
int32_t ret = 0;
for (int i = 0; i < taosArrayGetSize(offsets); i++) {
uint64_t offset = *(uint64_t*)taosArrayGet(offsets, i);
ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
if (ret != 0) {
indexError("failed to find target tablelist");
return TSDB_CODE_TDB_FILE_CORRUPTED;
}
}
if (hasJson) {
taosMemoryFree(p);
}
return 0;
}
static int32_t tfSearchSuffix(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
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);
}
int64_t st = taosGetTimestampUs();
FstSlice key = fstSliceCreate(p, sz);
/*impl later*/
if (hasJson) {
taosMemoryFree(p);
}
fstSliceDestroy(&key);
return 0;
}
static int32_t tfSearchRegex(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
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);
}
int64_t st = taosGetTimestampUs();
FstSlice key = fstSliceCreate(p, sz);
/*impl later*/
if (hasJson) {
taosMemoryFree(p);
}
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;
char* p = tem->colVal;
uint64_t sz = tem->nColVal;
if (hasJson) {
p = indexPackJsonData(tem);
sz = strlen(p);
}
int64_t st = taosGetTimestampUs();
FstSlice key = fstSliceCreate(p, sz);
// uint64_t offset;
// if (fstGet(((TFileReader*)reader)->fst, &key, &offset)) {
// int64_t et = taosGetTimestampUs();
// int64_t cost = et - st;
// indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
// tem->suid, tem->colName, tem->colVal, cost);
// ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
// cost = taosGetTimestampUs() - et;
// indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
// tem->colName, tem->colVal, cost);
//}
if (hasJson) {
taosMemoryFree(p);
}
fstSliceDestroy(&key);
return 0;
}
int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTempResult* tr) {
SIndexTerm* term = query->term;
bool hasJson = INDEX_TYPE_CONTAIN_EXTERN_TYPE(term->colType, TSDB_DATA_TYPE_JSON);
EIndexQueryType qtype = query->qType;
// SArray* result = taosArrayInit(16, sizeof(uint64_t));
int ret = -1;
// refactor to callback later
if (qtype == QUERY_TERM) {
uint64_t offset;
char* p = term->colVal;
uint64_t sz = term->nColVal;
if (hasJson) {
p = indexPackJsonData(term);
sz = strlen(p);
}
int64_t st = taosGetTimestampUs();
FstSlice key = fstSliceCreate(p, sz);
if (fstGet(reader->fst, &key, &offset)) {
int64_t et = taosGetTimestampUs();
int64_t cost = et - st;
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
term->suid, term->colName, term->colVal, cost);
ret = tfileReaderLoadTableIds(reader, offset, tr->total);
cost = taosGetTimestampUs() - et;
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", term->suid,
term->colName, term->colVal, cost);
} else {
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName,
term->colVal);
}
fstSliceDestroy(&key);
if (hasJson) {
taosMemoryFree(p);
}
} else if (qtype == QUERY_PREFIX) {
// handle later
//
} else if (qtype == QUERY_SUFFIX) {
// handle later
} else if (qtype == QUERY_REGEX) {
// handle later
} else if (qtype == QUERY_RANGE) {
// handle later
if (qtype >= sizeof(tfSearch) / sizeof(tfSearch[0])) {
indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, not found table info in tindex", term->suid, term->colName,
term->colVal);
return -1;
} else {
return tfSearch[qtype](reader, term, tr);
}
tfileReaderUnRef(reader);
// taosArrayAddAll(tr->total, result);
// taosArrayDestroy(result);
return ret;
return 0;
}
TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int32_t version, const char* colName, uint8_t colType) {

50
source/libs/index/test/indexTests.cc
View File

@ -483,9 +483,9 @@ TEST_F(IndexTFileEnv, test_tfile_write) {
std::string colName("voltage");
std::string colVal("ab");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexTermQuery query = { term, QUERY_TERM};
SIndexTermQuery query = {term, QUERY_TERM};
SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t));
fObj->Get(&query, result);
@ -557,7 +557,7 @@ TEST_F(IndexCacheEnv, cache_test) {
std::string colName("voltage");
{
std::string colVal("v1");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
@ -565,28 +565,28 @@ TEST_F(IndexCacheEnv, cache_test) {
}
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
}
{
std::string colVal("v2");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
}
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
}
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
@ -595,14 +595,14 @@ TEST_F(IndexCacheEnv, cache_test) {
std::cout << "--------first----------" << std::endl;
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, othColId, version++, suid++);
indexTermDestroy(term);
}
{
std::string colVal("v4");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, othColId, version++, suid++);
indexTermDestroy(term);
@ -613,7 +613,7 @@ TEST_F(IndexCacheEnv, cache_test) {
std::string colVal("v4");
for (size_t i = 0; i < 10; i++) {
colVal[colVal.size() - 1] = 'a' + i;
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
coj->Put(term, colId, version++, suid++);
indexTermDestroy(term);
@ -623,9 +623,9 @@ TEST_F(IndexCacheEnv, cache_test) {
// begin query
{
std::string colVal("v3");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexTermQuery query = { term, QUERY_TERM };
SIndexTermQuery query = {term, QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
@ -638,9 +638,9 @@ TEST_F(IndexCacheEnv, cache_test) {
}
{
std::string colVal("v2");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexTermQuery query = { term, QUERY_TERM };
SIndexTermQuery query = {term, QUERY_TERM};
SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid));
STermValueType valType;
@ -670,7 +670,7 @@ class IndexObj {
return ret;
}
void Del(const std::string& colName, const std::string& colVal, uint64_t uid) {
SIndexTerm* term = indexTermCreate(0, DEL_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, DEL_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -679,7 +679,7 @@ class IndexObj {
}
int WriteMillonData(const std::string& colName, const std::string& colVal = "Hello world",
size_t numOfTable = 100 * 10000) {
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -701,7 +701,7 @@ class IndexObj {
// opt
tColVal[taosRand() % colValSize] = 'a' + k % 26;
}
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
tColVal.c_str(), tColVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -737,7 +737,7 @@ class IndexObj {
int SearchOne(const std::string& colName, const std::string& colVal) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
@ -759,7 +759,7 @@ class IndexObj {
}
int SearchOneTarget(const std::string& colName, const std::string& colVal, uint64_t val) {
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);
@ -784,7 +784,7 @@ class IndexObj {
void PutOne(const std::string& colName, const std::string& colVal) {
SIndexMultiTerm* terms = indexMultiTermCreate();
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
Put(terms, 10);
@ -792,7 +792,7 @@ class IndexObj {
}
void PutOneTarge(const std::string& colName, const std::string& colVal, uint64_t val) {
SIndexMultiTerm* terms = indexMultiTermCreate();
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
Put(terms, val);
@ -832,7 +832,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
{
std::string colName("tag1"), colVal("Hello");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -847,7 +847,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
size_t size = 200;
std::string colName("tag1"), colVal("hello");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -862,7 +862,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
size_t size = 200;
std::string colName("tag1"), colVal("Hello");
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
@ -877,7 +877,7 @@ TEST_F(IndexEnv2, testIndexOpen) {
{
std::string colName("tag1"), colVal("Hello");
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(0, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
indexMultiTermQueryAdd(mq, term, QUERY_TERM);

16
source/libs/index/test/jsonUT.cc
View File

@ -40,7 +40,7 @@ TEST_F(JsonEnv, testWrite) {
{
std::string colName("test");
std::string colVal("ab");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -53,7 +53,7 @@ TEST_F(JsonEnv, testWrite) {
{
std::string colName("voltage");
std::string colVal("ab1");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -66,7 +66,7 @@ TEST_F(JsonEnv, testWrite) {
{
std::string colName("voltage");
std::string colVal("123");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -81,7 +81,7 @@ TEST_F(JsonEnv, testWrite) {
std::string colVal("ab");
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* q = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));
@ -95,7 +95,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
{
std::string colName("test");
std::string colVal("ab");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -110,7 +110,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
std::string colVal("abxxxxxxxxxxxx");
for (int i = 0; i < 1000; i++) {
colVal[i % colVal.size()] = '0' + i % 128;
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -124,7 +124,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
{
std::string colName("voltagefdadfa");
std::string colVal("abxxxxxxxxxxxx");
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* term = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SIndexMultiTerm* terms = indexMultiTermCreate();
@ -139,7 +139,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
std::string colVal("ab");
SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST);
SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
SIndexTerm* q = indexTermCreate(1, ADD_VALUE, 0, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size());
SArray* result = taosArrayInit(1, sizeof(uint64_t));

3
source/util/src/terror.c
View File

@ -271,7 +271,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_FUNC_RETRIEVE, "Invalid func retrieve
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_ALREADY_EXIST, "Transaction already exists")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_NOT_EXIST, "Transaction not exists")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_INVALID_STAGE, "Invalid stage to kill")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_CANT_PARALLEL, "Invalid stage to kill")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_CONFLICT, "Conflict transaction not completed")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_UNKNOW_ERROR, "Unknown transaction error")
// mnode-mq
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST, "Topic already exists")