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

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

This commit is contained in:
gccgdb1234 2022-05-31 17:30:17 +08:00
parent aa6569fa69 691cc93ed2
commit e68772df00
26 changed files with 786 additions and 488 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/util/tdef.h
View File

@ -254,6 +254,7 @@ typedef enum ELogicConditionType {
#define TSDB_TRANS_STAGE_LEN 12 #define TSDB_TRANS_STAGE_LEN 12
#define TSDB_TRANS_TYPE_LEN 16 #define TSDB_TRANS_TYPE_LEN 16
#define TSDB_TRANS_ERROR_LEN 64 #define TSDB_TRANS_ERROR_LEN 64
#define TSDB_TRANS_DESC_LEN 128
#define TSDB_STEP_NAME_LEN 32 #define TSDB_STEP_NAME_LEN 32
#define TSDB_STEP_DESC_LEN 128 #define TSDB_STEP_DESC_LEN 128

2
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -130,7 +130,7 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
_OVER: _OVER:
if (code != 0) { if (code != 0) {
dError("msg:%p, failed to process since %s", pMsg, terrstr()); dTrace("msg:%p, failed to process since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pRpc->msgType));
if (terrno != 0) code = terrno; if (terrno != 0) code = terrno;
if (IsReq(pRpc)) { if (IsReq(pRpc)) {

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

@ -60,14 +60,12 @@ typedef enum {
typedef enum { typedef enum {
TRN_STAGE_PREPARE = 0, TRN_STAGE_PREPARE = 0,
TRN_STAGE_REDO_LOG = 1, TRN_STAGE_REDO_ACTION = 1,
TRN_STAGE_REDO_ACTION = 2, TRN_STAGE_ROLLBACK = 2,
TRN_STAGE_ROLLBACK = 3, TRN_STAGE_UNDO_ACTION = 3,
TRN_STAGE_UNDO_ACTION = 4, TRN_STAGE_COMMIT = 4,
TRN_STAGE_UNDO_LOG = 5, TRN_STAGE_COMMIT_ACTION = 5,
TRN_STAGE_COMMIT = 6, TRN_STAGE_FINISHED = 6
TRN_STAGE_COMMIT_LOG = 7,
TRN_STAGE_FINISHED = 8
} ETrnStage; } ETrnStage;
typedef enum { typedef enum {
@ -131,7 +129,7 @@ typedef enum {
typedef enum { typedef enum {
TRN_EXEC_PARALLEL = 0, TRN_EXEC_PARALLEL = 0,
TRN_EXEC_ONE_BY_ONE = 1, TRN_EXEC_NO_PARALLEL = 1,
} ETrnExecType; } ETrnExecType;
typedef enum { typedef enum {
@ -168,16 +166,16 @@ typedef struct {
SRpcHandleInfo rpcInfo; SRpcHandleInfo rpcInfo;
void* rpcRsp; void* rpcRsp;
int32_t rpcRspLen; int32_t rpcRspLen;
SArray* redoLogs; int32_t redoActionPos;
SArray* undoLogs;
SArray* commitLogs;
SArray* redoActions; SArray* redoActions;
SArray* undoActions; SArray* undoActions;
SArray* commitActions;
int64_t createdTime; int64_t createdTime;
int64_t lastExecTime; int64_t lastExecTime;
int64_t dbUid; int64_t dbUid;
char dbname[TSDB_DB_FNAME_LEN]; char dbname[TSDB_DB_FNAME_LEN];
char lastError[TSDB_TRANS_ERROR_LEN]; char lastError[TSDB_TRANS_ERROR_LEN];
char desc[TSDB_TRANS_DESC_LEN];
int32_t startFunc; int32_t startFunc;
int32_t stopFunc; int32_t stopFunc;
int32_t paramLen; int32_t paramLen;

37
source/dnode/mnode/impl/inc/mndTrans.h
View File

@ -26,31 +26,24 @@ typedef enum {
TRANS_START_FUNC_TEST = 1, TRANS_START_FUNC_TEST = 1,
TRANS_STOP_FUNC_TEST = 2, TRANS_STOP_FUNC_TEST = 2,
TRANS_START_FUNC_MQ_REB = 3, TRANS_START_FUNC_MQ_REB = 3,
TRANS_STOP_FUNC_TEST_MQ_REB = 4, TRANS_STOP_FUNC_MQ_REB = 4,
} ETrnFunc; } ETrnFunc;
typedef struct { typedef struct {
SEpSet epSet; int32_t id;
tmsg_t msgType; int32_t errCode;
int8_t msgSent; int32_t acceptableCode;
int8_t msgReceived; int8_t stage;
int32_t errCode; int8_t isRaw;
int32_t acceptableCode; int8_t rawWritten;
int32_t contLen; int8_t msgSent;
void *pCont; int8_t msgReceived;
} STransAction; tmsg_t msgType;
SEpSet epSet;
typedef struct { int32_t contLen;
void *pCont;
SSdbRaw *pRaw; SSdbRaw *pRaw;
} STransLog; } STransAction;
typedef struct {
ETrnStep stepType;
STransAction redoAction;
STransAction undoAction;
STransLog redoLog;
STransLog undoLog;
} STransStep;
typedef void (*TransCbFp)(SMnode *pMnode, void *param, int32_t paramLen); typedef void (*TransCbFp)(SMnode *pMnode, void *param, int32_t paramLen);
@ -69,7 +62,7 @@ int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction);
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen); void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen);
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen); void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen);
void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb); void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
void mndTransSetExecOneByOne(STrans *pTrans); void mndTransSetNoParallel(STrans *pTrans);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans); int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransProcessRsp(SRpcMsg *pRsp); void mndTransProcessRsp(SRpcMsg *pRsp);

8
source/dnode/mnode/impl/src/mndAcct.c
View File

@ -78,10 +78,8 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
if (pRaw == NULL) return -1; if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY); sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mDebug("acct:%s, will be created while deploy sdb, raw:%p", acctObj.acct, pRaw); mDebug("acct:%s, will be created when deploying, raw:%p", acctObj.acct, pRaw);
#if 0
return sdbWrite(pMnode->pSdb, pRaw);
#else
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_ACCT, NULL); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_ACCT, NULL);
if (pTrans == NULL) { if (pTrans == NULL) {
mError("acct:%s, failed to create since %s", acctObj.acct, terrstr()); mError("acct:%s, failed to create since %s", acctObj.acct, terrstr());
@ -94,7 +92,6 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
} }
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
if (mndTransPrepare(pMnode, pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@ -104,7 +101,6 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
mndTransDrop(pTrans); mndTransDrop(pTrans);
return 0; return 0;
#endif
} }
static SSdbRaw *mndAcctActionEncode(SAcctObj *pAcct) { static SSdbRaw *mndAcctActionEncode(SAcctObj *pAcct) {

4
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -172,13 +172,13 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
clusterObj.id = mndGenerateUid(clusterObj.name, TSDB_CLUSTER_ID_LEN); clusterObj.id = mndGenerateUid(clusterObj.name, TSDB_CLUSTER_ID_LEN);
clusterObj.id = (clusterObj.id >= 0 ? clusterObj.id : -clusterObj.id); clusterObj.id = (clusterObj.id >= 0 ? clusterObj.id : -clusterObj.id);
pMnode->clusterId = clusterObj.id; pMnode->clusterId = clusterObj.id;
mDebug("cluster:%" PRId64 ", name is %s", clusterObj.id, clusterObj.name); mInfo("cluster:%" PRId64 ", name is %s", clusterObj.id, clusterObj.name);
SSdbRaw *pRaw = mndClusterActionEncode(&clusterObj); SSdbRaw *pRaw = mndClusterActionEncode(&clusterObj);
if (pRaw == NULL) return -1; if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY); sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mDebug("cluster:%" PRId64 ", will be created while deploy sdb, raw:%p", clusterObj.id, pRaw); mDebug("cluster:%" PRId64 ", will be created when deploying, raw:%p", clusterObj.id, pRaw);
#if 0 #if 0
return sdbWrite(pMnode->pSdb, pRaw); return sdbWrite(pMnode->pSdb, pRaw);
#else #else

2
source/dnode/mnode/impl/src/mndDb.c
View File

@ -1314,7 +1314,7 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
SDbObj *pDb = mndAcquireDb(pMnode, pDbVgVersion->dbFName); SDbObj *pDb = mndAcquireDb(pMnode, pDbVgVersion->dbFName);
if (pDb == NULL) { if (pDb == NULL) {
mDebug("db:%s, no exist", pDbVgVersion->dbFName); mTrace("db:%s, no exist", pDbVgVersion->dbFName);
memcpy(usedbRsp.db, pDbVgVersion->dbFName, TSDB_DB_FNAME_LEN); memcpy(usedbRsp.db, pDbVgVersion->dbFName, TSDB_DB_FNAME_LEN);
usedbRsp.uid = pDbVgVersion->dbId; usedbRsp.uid = pDbVgVersion->dbId;
usedbRsp.vgVersion = -1; usedbRsp.vgVersion = -1;

10
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -98,7 +98,7 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
if (pRaw == NULL) return -1; if (pRaw == NULL) return -1;
if (sdbSetRawStatus(pRaw, SDB_STATUS_READY) != 0) return -1; if (sdbSetRawStatus(pRaw, SDB_STATUS_READY) != 0) return -1;
mDebug("dnode:%d, will be created while deploy sdb, raw:%p", dnodeObj.id, pRaw); mDebug("dnode:%d, will be created when deploying, raw:%p", dnodeObj.id, pRaw);
#if 0 #if 0
return sdbWrite(pMnode->pSdb, pRaw); return sdbWrite(pMnode->pSdb, pRaw);
@ -388,9 +388,10 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
mndReleaseMnode(pMnode, pObj); mndReleaseMnode(pMnode, pObj);
} }
int64_t dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE) + sdbGetTableVer(pMnode->pSdb, SDB_MNODE);
int64_t curMs = taosGetTimestampMs(); int64_t curMs = taosGetTimestampMs();
bool online = mndIsDnodeOnline(pMnode, pDnode, curMs); bool online = mndIsDnodeOnline(pMnode, pDnode, curMs);
bool dnodeChanged = (statusReq.dnodeVer != sdbGetTableVer(pMnode->pSdb, SDB_DNODE)); bool dnodeChanged = (statusReq.dnodeVer != dnodeVer);
bool reboot = (pDnode->rebootTime != statusReq.rebootTime); bool reboot = (pDnode->rebootTime != statusReq.rebootTime);
bool needCheck = !online || dnodeChanged || reboot; bool needCheck = !online || dnodeChanged || reboot;
@ -433,7 +434,8 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
if (!online) { if (!online) {
mInfo("dnode:%d, from offline to online", pDnode->id); mInfo("dnode:%d, from offline to online", pDnode->id);
} else { } else {
mDebug("dnode:%d, send dnode eps", pDnode->id); mDebug("dnode:%d, send dnode epset, online:%d ver:% " PRId64 ":%" PRId64 " reboot:%d", pDnode->id, online,
statusReq.dnodeVer, dnodeVer, reboot);
} }
pDnode->rebootTime = statusReq.rebootTime; pDnode->rebootTime = statusReq.rebootTime;
@ -441,7 +443,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes; pDnode->numOfSupportVnodes = statusReq.numOfSupportVnodes;
SStatusRsp statusRsp = {0}; SStatusRsp statusRsp = {0};
statusRsp.dnodeVer = sdbGetTableVer(pMnode->pSdb, SDB_DNODE) + sdbGetTableVer(pMnode->pSdb, SDB_MNODE); statusRsp.dnodeVer = dnodeVer;
statusRsp.dnodeCfg.dnodeId = pDnode->id; statusRsp.dnodeCfg.dnodeId = pDnode->id;
statusRsp.dnodeCfg.clusterId = pMnode->clusterId; statusRsp.dnodeCfg.clusterId = pMnode->clusterId;
statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp)); statusRsp.pDnodeEps = taosArrayInit(mndGetDnodeSize(pMnode), sizeof(SDnodeEp));

2
source/dnode/mnode/impl/src/mndMain.c
View File

@ -472,7 +472,7 @@ int32_t mndProcessRpcMsg(SRpcMsg *pMsg) {
} else if (code == 0) { } else if (code == 0) {
mTrace("msg:%p, successfully processed and response", pMsg); mTrace("msg:%p, successfully processed and response", pMsg);
} else { } else {
mError("msg:%p, failed to process since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle, mDebug("msg:%p, failed to process since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,
TMSG_INFO(pMsg->msgType)); TMSG_INFO(pMsg->msgType));
} }

6
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -90,7 +90,7 @@ static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
if (pRaw == NULL) return -1; if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY); sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mDebug("mnode:%d, will be created while deploy sdb, raw:%p", mnodeObj.id, pRaw); mDebug("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
#if 0 #if 0
return sdbWrite(pMnode->pSdb, pRaw); return sdbWrite(pMnode->pSdb, pRaw);
@ -367,7 +367,7 @@ static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
if (pTrans == NULL) goto _OVER; if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId); mDebug("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
mndTransSetExecOneByOne(pTrans); mndTransSetNoParallel(pTrans);
if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER; if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER; if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto _OVER; if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto _OVER;
@ -539,7 +539,7 @@ static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
if (pTrans == NULL) goto _OVER; if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id); mDebug("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
mndTransSetExecOneByOne(pTrans); mndTransSetNoParallel(pTrans);
if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) goto _OVER; if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) goto _OVER;
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) goto _OVER; if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) goto _OVER;
if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) goto _OVER; if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) goto _OVER;

2
source/dnode/mnode/impl/src/mndSma.c
View File

@ -507,7 +507,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
mDebug("trans:%d, used to create sma:%s", pTrans->id, pCreate->name); mDebug("trans:%d, used to create sma:%s", pTrans->id, pCreate->name);
mndTransSetDbInfo(pTrans, pDb); mndTransSetDbInfo(pTrans, pDb);
mndTransSetExecOneByOne(pTrans); mndTransSetNoParallel(pTrans);
if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER; if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER; if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;

2
source/dnode/mnode/impl/src/mndStb.c
View File

@ -1597,7 +1597,7 @@ static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
pReq->info.rspLen = rspLen; pReq->info.rspLen = rspLen;
code = 0; code = 0;
mDebug("stb:%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName); mTrace("%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
_OVER: _OVER:
if (code != 0) { if (code != 0) {

2
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -501,7 +501,7 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
// 4. TODO commit log: modification log // 4. TODO commit log: modification log
// 5. set cb // 5. set cb
mndTransSetCb(pTrans, TRANS_START_FUNC_MQ_REB, TRANS_STOP_FUNC_TEST_MQ_REB, NULL, 0); mndTransSetCb(pTrans, TRANS_START_FUNC_MQ_REB, TRANS_STOP_FUNC_MQ_REB, NULL, 0);
// 6. execution // 6. execution
if (mndTransPrepare(pMnode, pTrans) != 0) { if (mndTransPrepare(pMnode, pTrans) != 0) {

4
source/dnode/mnode/impl/src/mndSync.c
View File

@ -65,7 +65,7 @@ int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) {
void mndRestoreFinish(struct SSyncFSM *pFsm) { void mndRestoreFinish(struct SSyncFSM *pFsm) {
SMnode *pMnode = pFsm->data; SMnode *pMnode = pFsm->data;
if (!pMnode->deploy) { if (!pMnode->deploy) {
mInfo("mnode sync restore finished"); mInfo("mnode sync restore finished, and will handle outstanding transactions");
mndTransPullup(pMnode); mndTransPullup(pMnode);
mndSetRestore(pMnode, true); mndSetRestore(pMnode, true);
} else { } else {
@ -244,7 +244,7 @@ void mndSyncStart(SMnode *pMnode) {
} else { } else {
syncStart(pMgmt->sync); syncStart(pMgmt->sync);
} }
mDebug("sync:%" PRId64 " is started, standby:%d", pMgmt->sync, pMgmt->standby); mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby);
} }
void mndSyncStop(SMnode *pMnode) {} void mndSyncStop(SMnode *pMnode) {}

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

File diff suppressed because it is too large Load Diff

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

@ -77,7 +77,7 @@ static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char
if (pRaw == NULL) return -1; if (pRaw == NULL) return -1;
sdbSetRawStatus(pRaw, SDB_STATUS_READY); sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mDebug("user:%s, will be created while deploy sdb, raw:%p", userObj.user, pRaw); mDebug("user:%s, will be created when deploying, raw:%p", userObj.user, pRaw);
#if 0 #if 0
return sdbWrite(pMnode->pSdb, pRaw); return sdbWrite(pMnode->pSdb, pRaw);

4
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -501,7 +501,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
*ppVgroups = pVgroups; *ppVgroups = pVgroups;
code = 0; code = 0;
mInfo("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications); mInfo("db:%s, total %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
_OVER: _OVER:
if (code != 0) taosMemoryFree(pVgroups); if (code != 0) taosMemoryFree(pVgroups);
@ -539,7 +539,7 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
pVgid->role = TAOS_SYNC_STATE_FOLLOWER; pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
pDnode->numOfVnodes++; pDnode->numOfVnodes++;
mInfo("db:%s, vgId:%d, vn:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId); mInfo("db:%s, vgId:%d, vnode_index:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId);
maxPos++; maxPos++;
if (maxPos == 3) return 0; if (maxPos == 3) return 0;
} }

1
source/dnode/mnode/sdb/inc/sdb.h
View File

@ -168,6 +168,7 @@ typedef struct SSdb {
char *currDir; char *currDir;
char *tmpDir; char *tmpDir;
int64_t lastCommitVer; int64_t lastCommitVer;
int64_t lastCommitTerm;
int64_t curVer; int64_t curVer;
int64_t curTerm; int64_t curTerm;
int64_t tableVer[SDB_MAX]; int64_t tableVer[SDB_MAX];

1
source/dnode/mnode/sdb/src/sdb.c
View File

@ -55,6 +55,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
pSdb->curVer = -1; pSdb->curVer = -1;
pSdb->curTerm = -1; pSdb->curTerm = -1;
pSdb->lastCommitVer = -1; pSdb->lastCommitVer = -1;
pSdb->lastCommitTerm = -1;
pSdb->pMnode = pOption->pMnode; pSdb->pMnode = pOption->pMnode;
taosThreadMutexInit(&pSdb->filelock, NULL); taosThreadMutexInit(&pSdb->filelock, NULL);
mDebug("sdb init successfully"); mDebug("sdb init successfully");

54
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -70,6 +70,7 @@ static void sdbResetData(SSdb *pSdb) {
pSdb->curVer = -1; pSdb->curVer = -1;
pSdb->curTerm = -1; pSdb->curTerm = -1;
pSdb->lastCommitVer = -1; pSdb->lastCommitVer = -1;
pSdb->lastCommitTerm = -1;
mDebug("sdb reset successfully"); mDebug("sdb reset successfully");
} }
@ -211,12 +212,12 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
char file[PATH_MAX] = {0}; char file[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP); snprintf(file, sizeof(file), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);
mDebug("start to read file:%s", file); mDebug("start to read sdb file:%s", file);
SSdbRaw *pRaw = taosMemoryMalloc(WAL_MAX_SIZE + 100); SSdbRaw *pRaw = taosMemoryMalloc(WAL_MAX_SIZE + 100);
if (pRaw == NULL) { if (pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("failed read file since %s", terrstr()); mError("failed read sdb file since %s", terrstr());
return -1; return -1;
} }
@ -224,12 +225,12 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
if (pFile == NULL) { if (pFile == NULL) {
taosMemoryFree(pRaw); taosMemoryFree(pRaw);
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to read file:%s since %s", file, terrstr()); mError("failed to read sdb file:%s since %s", file, terrstr());
return 0; return 0;
} }
if (sdbReadFileHead(pSdb, pFile) != 0) { if (sdbReadFileHead(pSdb, pFile) != 0) {
mError("failed to read file:%s head since %s", file, terrstr()); mError("failed to read sdb file:%s head since %s", file, terrstr());
taosMemoryFree(pRaw); taosMemoryFree(pRaw);
taosCloseFile(&pFile); taosCloseFile(&pFile);
return -1; return -1;
@ -245,13 +246,13 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
if (ret < 0) { if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
mError("failed to read file:%s since %s", file, tstrerror(code)); mError("failed to read sdb file:%s since %s", file, tstrerror(code));
break; break;
} }
if (ret != readLen) { if (ret != readLen) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
mError("failed to read file:%s since %s", file, tstrerror(code)); mError("failed to read sdb file:%s since %s", file, tstrerror(code));
break; break;
} }
@ -259,34 +260,36 @@ static int32_t sdbReadFileImp(SSdb *pSdb) {
ret = taosReadFile(pFile, pRaw->pData, readLen); ret = taosReadFile(pFile, pRaw->pData, readLen);
if (ret < 0) { if (ret < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
mError("failed to read file:%s since %s", file, tstrerror(code)); mError("failed to read sdb file:%s since %s", file, tstrerror(code));
break; break;
} }
if (ret != readLen) { if (ret != readLen) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
mError("failed to read file:%s since %s", file, tstrerror(code)); mError("failed to read sdb file:%s since %s", file, tstrerror(code));
break; break;
} }
int32_t totalLen = sizeof(SSdbRaw) + pRaw->dataLen + sizeof(int32_t); int32_t totalLen = sizeof(SSdbRaw) + pRaw->dataLen + sizeof(int32_t);
if ((!taosCheckChecksumWhole((const uint8_t *)pRaw, totalLen)) != 0) { if ((!taosCheckChecksumWhole((const uint8_t *)pRaw, totalLen)) != 0) {
code = TSDB_CODE_CHECKSUM_ERROR; code = TSDB_CODE_CHECKSUM_ERROR;
mError("failed to read file:%s since %s", file, tstrerror(code)); mError("failed to read sdb file:%s since %s", file, tstrerror(code));
break; break;
} }
code = sdbWriteWithoutFree(pSdb, pRaw); code = sdbWriteWithoutFree(pSdb, pRaw);
if (code != 0) { if (code != 0) {
mError("failed to read file:%s since %s", file, terrstr()); mError("failed to read sdb file:%s since %s", file, terrstr());
goto _OVER; goto _OVER;
} }
} }
code = 0; code = 0;
pSdb->lastCommitVer = pSdb->curVer; pSdb->lastCommitVer = pSdb->curVer;
pSdb->lastCommitTerm = pSdb->curTerm;
memcpy(pSdb->tableVer, tableVer, sizeof(tableVer)); memcpy(pSdb->tableVer, tableVer, sizeof(tableVer));
mDebug("read file:%s successfully, ver:%" PRId64, file, pSdb->lastCommitVer); mDebug("read sdb file:%s successfully, ver:%" PRId64 " term:%" PRId64, file, pSdb->lastCommitVer,
pSdb->lastCommitTerm);
_OVER: _OVER:
taosCloseFile(&pFile); taosCloseFile(&pFile);
@ -302,7 +305,7 @@ int32_t sdbReadFile(SSdb *pSdb) {
sdbResetData(pSdb); sdbResetData(pSdb);
int32_t code = sdbReadFileImp(pSdb); int32_t code = sdbReadFileImp(pSdb);
if (code != 0) { if (code != 0) {
mError("failed to read sdb since %s", terrstr()); mError("failed to read sdb file since %s", terrstr());
sdbResetData(pSdb); sdbResetData(pSdb);
} }
@ -318,18 +321,19 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
char curfile[PATH_MAX] = {0}; char curfile[PATH_MAX] = {0};
snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP); snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);
mDebug("start to write file:%s, current ver:%" PRId64 " term:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer, mDebug("start to write sdb file, current ver:%" PRId64 " term:%" PRId64 ", commit ver:%" PRId64 " term:%" PRId64
pSdb->curTerm, pSdb->lastCommitVer); " file:%s",
pSdb->curVer, pSdb->curTerm, pSdb->lastCommitVer, pSdb->lastCommitTerm, curfile);
TdFilePtr pFile = taosOpenFile(tmpfile, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); TdFilePtr pFile = taosOpenFile(tmpfile, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) { if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to open file:%s for write since %s", tmpfile, terrstr()); mError("failed to open sdb file:%s for write since %s", tmpfile, terrstr());
return -1; return -1;
} }
if (sdbWriteFileHead(pSdb, pFile) != 0) { if (sdbWriteFileHead(pSdb, pFile) != 0) {
mError("failed to write file:%s head since %s", tmpfile, terrstr()); mError("failed to write sdb file:%s head since %s", tmpfile, terrstr());
taosCloseFile(&pFile); taosCloseFile(&pFile);
return -1; return -1;
} }
@ -338,7 +342,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
SdbEncodeFp encodeFp = pSdb->encodeFps[i]; SdbEncodeFp encodeFp = pSdb->encodeFps[i];
if (encodeFp == NULL) continue; if (encodeFp == NULL) continue;
mTrace("write %s to file, total %d rows", sdbTableName(i), sdbGetSize(pSdb, i)); mTrace("write %s to sdb file, total %d rows", sdbTableName(i), sdbGetSize(pSdb, i));
SHashObj *hash = pSdb->hashObjs[i]; SHashObj *hash = pSdb->hashObjs[i];
TdThreadRwlock *pLock = &pSdb->locks[i]; TdThreadRwlock *pLock = &pSdb->locks[i];
@ -394,7 +398,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
code = taosFsyncFile(pFile); code = taosFsyncFile(pFile);
if (code != 0) { if (code != 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
mError("failed to sync file:%s since %s", tmpfile, tstrerror(code)); mError("failed to sync sdb file:%s since %s", tmpfile, tstrerror(code));
} }
} }
@ -404,15 +408,17 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
code = taosRenameFile(tmpfile, curfile); code = taosRenameFile(tmpfile, curfile);
if (code != 0) { if (code != 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
mError("failed to write file:%s since %s", curfile, tstrerror(code)); mError("failed to write sdb file:%s since %s", curfile, tstrerror(code));
} }
} }
if (code != 0) { if (code != 0) {
mError("failed to write file:%s since %s", curfile, tstrerror(code)); mError("failed to write sdb file:%s since %s", curfile, tstrerror(code));
} else { } else {
pSdb->lastCommitVer = pSdb->curVer; pSdb->lastCommitVer = pSdb->curVer;
mDebug("write file:%s successfully, ver:%" PRId64 " term:%" PRId64, curfile, pSdb->lastCommitVer, pSdb->curTerm); pSdb->lastCommitTerm = pSdb->curTerm;
mDebug("write sdb file successfully, ver:%" PRId64 " term:%" PRId64 " file:%s", pSdb->lastCommitVer,
pSdb->lastCommitTerm, curfile);
} }
terrno = code; terrno = code;
@ -427,7 +433,7 @@ int32_t sdbWriteFile(SSdb *pSdb) {
taosThreadMutexLock(&pSdb->filelock); taosThreadMutexLock(&pSdb->filelock);
int32_t code = sdbWriteFileImp(pSdb); int32_t code = sdbWriteFileImp(pSdb);
if (code != 0) { if (code != 0) {
mError("failed to write sdb since %s", terrstr()); mError("failed to write sdb file since %s", terrstr());
} }
taosThreadMutexUnlock(&pSdb->filelock); taosThreadMutexUnlock(&pSdb->filelock);
return code; return code;
@ -493,7 +499,7 @@ int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) {
if (taosCopyFile(datafile, pIter->name) < 0) { if (taosCopyFile(datafile, pIter->name) < 0) {
taosThreadMutexUnlock(&pSdb->filelock); taosThreadMutexUnlock(&pSdb->filelock);
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to copy file %s to %s since %s", datafile, pIter->name, terrstr()); mError("failed to copy sdb file %s to %s since %s", datafile, pIter->name, terrstr());
sdbCloseIter(pIter); sdbCloseIter(pIter);
return -1; return -1;
} }
@ -502,7 +508,7 @@ int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter) {
pIter->file = taosOpenFile(pIter->name, TD_FILE_READ); pIter->file = taosOpenFile(pIter->name, TD_FILE_READ);
if (pIter->file == NULL) { if (pIter->file == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
mError("failed to open file:%s since %s", pIter->name, terrstr()); mError("failed to open sdb file:%s since %s", pIter->name, terrstr());
sdbCloseIter(pIter); sdbCloseIter(pIter);
return -1; return -1;
} }

6
source/libs/qworker/src/qworker.c
View File

@ -79,7 +79,11 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
if (taskHandle) { if (taskHandle) {
code = qExecTask(taskHandle, &pRes, &useconds); code = qExecTask(taskHandle, &pRes, &useconds);
if (code) { if (code) {
QW_TASK_ELOG("qExecTask failed, code:%x - %s", code, tstrerror(code)); if (code != TSDB_CODE_OPS_NOT_SUPPORT) {
QW_TASK_ELOG("qExecTask failed, code:%x - %s", code, tstrerror(code));
} else {
QW_TASK_DLOG("qExecTask failed, code:%x - %s", code, tstrerror(code));
}
QW_ERR_RET(code); QW_ERR_RET(code);
} }
} }

38
tests/pytest/util/dnodes.py
View File

@ -494,6 +494,7 @@ class TDDnodes:
self.simDeployed = False self.simDeployed = False
self.testCluster = False self.testCluster = False
self.valgrind = 0 self.valgrind = 0
self.killValgrind = 1
def init(self, path, remoteIP = ""): def init(self, path, remoteIP = ""):
psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}'" psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}'"
@ -505,14 +506,15 @@ class TDDnodes:
processID = subprocess.check_output( processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8") psCmd, shell=True).decode("utf-8")
psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}'" if self.killValgrind == 1:
processID = subprocess.check_output(psCmd, shell=True).decode("utf-8") psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}'"
while(processID): processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
killCmd = "kill -9 %s > /dev/null 2>&1" % processID while(processID):
os.system(killCmd) killCmd = "kill -9 %s > /dev/null 2>&1" % processID
time.sleep(1) os.system(killCmd)
processID = subprocess.check_output( time.sleep(1)
psCmd, shell=True).decode("utf-8") processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
binPath = self.dnodes[0].getPath() + "/../../../" binPath = self.dnodes[0].getPath() + "/../../../"
# tdLog.debug("binPath %s" % (binPath)) # tdLog.debug("binPath %s" % (binPath))
@ -549,6 +551,9 @@ class TDDnodes:
def setValgrind(self, value): def setValgrind(self, value):
self.valgrind = value self.valgrind = value
def setKillValgrind(self, value):
self.killValgrind = value
def deploy(self, index, *updatecfgDict): def deploy(self, index, *updatecfgDict):
self.sim.setTestCluster(self.testCluster) self.sim.setTestCluster(self.testCluster)
@ -622,14 +627,15 @@ class TDDnodes:
processID = subprocess.check_output( processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8") psCmd, shell=True).decode("utf-8")
psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}'" if self.killValgrind == 1:
processID = subprocess.check_output(psCmd, shell=True).decode("utf-8") psCmd = "ps -ef|grep -w valgrind.bin| grep -v grep | awk '{print $2}'"
while(processID): processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
killCmd = "kill -TERM %s > /dev/null 2>&1" % processID while(processID):
os.system(killCmd) killCmd = "kill -TERM %s > /dev/null 2>&1" % processID
time.sleep(1) os.system(killCmd)
processID = subprocess.check_output( time.sleep(1)
psCmd, shell=True).decode("utf-8") processID = subprocess.check_output(
psCmd, shell=True).decode("utf-8")
# if os.system(cmd) != 0 : # if os.system(cmd) != 0 :
# tdLog.exit(cmd) # tdLog.exit(cmd)

265
tests/system-test/2-query/stateduration.py Normal file
View File

@ -0,0 +1,265 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.ts = 1537146000000
self.param_list = ['LT','lt','Lt','lT','GT','gt','Gt','gT','LE','le','Le','lE','GE','ge','Ge','gE','NE','ne','Ne','nE','EQ','eq','Eq','eQ']
self.row_num = 10
def run(self):
tdSql.prepare()
# timestamp = 1ms , time_unit = 1s
tdSql.execute('''create table test(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')
for i in range(self.row_num):
tdSql.execute("insert into test values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1))
integer_list = [1,2,3,4,11,12,13,14]
float_list = [5,6]
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5) from test")
tdSql.checkRows(10)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (0,), (0,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (0,), (0,), (0,), (0,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (0,), (0,), (0,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (0,), (0,), (0,), (0,), (0,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (0,), (0,), (0,), (-1,), (0,), (0,), (0,), (0,), (0,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5) from test")
tdSql.checkRows(10)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (0,), (0,), (0,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (0,), (0,), (0,), (0,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (0,), (0,), (0,), (0,), (0,), (0,), (0,), (0,), (0,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
error_column_list = ['ts','col7','col8','col9','a',1]
for i in error_column_list:
for j in self.param_list:
tdSql.error(f"select stateduration({i},{j},5) from test")
error_param_list = ['a',1]
for i in error_param_list:
tdSql.error(f"select stateduration(col1,{i},5) from test")
# timestamp = 1s, time_unit =1s
tdSql.execute('''create table test1(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')
for i in range(self.row_num):
tdSql.execute("insert into test1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i*1000, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1))
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5) from test1")
tdSql.checkRows(10)
# print(tdSql.queryResult)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,), (5,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5) from test1")
tdSql.checkRows(10)
print(tdSql.queryResult)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
# timestamp = 1m, time_unit =1m
tdSql.execute('''create table test2(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')
for i in range(self.row_num):
tdSql.execute("insert into test2 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i*1000*60, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1))
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1m) from test2")
tdSql.checkRows(10)
# print(tdSql.queryResult)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,), (5,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1m) from test2")
tdSql.checkRows(10)
print(tdSql.queryResult)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
# timestamp = 1h, time_unit =1h
tdSql.execute('''create table test3(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')
for i in range(self.row_num):
tdSql.execute("insert into test3 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i*1000*60*60, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1))
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1h) from test3")
tdSql.checkRows(10)
# print(tdSql.queryResult)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,), (5,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1h) from test3")
tdSql.checkRows(10)
print(tdSql.queryResult)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
# timestamp = 1h,time_unit =1m
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1m) from test3")
tdSql.checkRows(10)
# print(tdSql.queryResult)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (60,), (120,), (180,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (60,), (120,), (180,), (240,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (60,), (120,), (180,), (240,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (60,), (120,), (180,), (240,), (300,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (60,), (120,), (180,), (-1,), (0,), (60,), (120,), (180,), (240,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1m) from test3")
tdSql.checkRows(10)
print(tdSql.queryResult)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (60,), (120,), (180,), (240,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (60,), (120,), (180,), (240,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (60,), (120,), (180,), (240,), (300,), (360,), (420,), (480,), (540,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
# for stb
tdSql.execute('''create table stb(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned) tags(t0 int)''')
tdSql.execute('create table stb_1 using stb tags(1)')
for i in range(self.row_num):
tdSql.execute("insert into stb_1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i*1000*60*60, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1))
for i in integer_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1h) from stb")
tdSql.checkRows(10)
# print(tdSql.queryResult)
if j in ['LT' ,'lt','Lt','lT']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GT','gt', 'Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in [ 'GE','ge','Ge','gE']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,), (5,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (0,), (-1,), (-1,), (-1,), (-1,), (-1,)])
for i in float_list:
for j in self.param_list:
tdSql.query(f"select stateduration(col{i},'{j}',5,1h) from stb")
tdSql.checkRows(10)
print(tdSql.queryResult)
if j in ['LT','lt','Lt','lT','LE','le','Le','lE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (-1,), (-1,), (-1,), (-1,), (-1,)])
elif j in ['GE','ge','Ge','gE','GT','gt','Gt','gT']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (0,), (1,), (2,), (3,), (4,)])
elif j in ['NE','ne','Ne','nE']:
tdSql.checkEqual(tdSql.queryResult,[(0,), (1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,)])
elif j in ['EQ','eq','Eq','eQ']:
tdSql.checkEqual(tdSql.queryResult,[(-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,), (-1,)])
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

5
tests/system-test/fulltest.sh
View File

@ -55,8 +55,8 @@ python3 ./test.py -f 2-query/Timediff.py
python3 ./test.py -f 2-query/top.py python3 ./test.py -f 2-query/top.py
python3 ./test.py -f 2-query/bottom.py python3 ./test.py -f 2-query/bottom.py
python3 ./test.py -f 2-query/percentile.py
python3 ./test.py -f 2-query/apercentile.py
python3 ./test.py -f 2-query/abs.py python3 ./test.py -f 2-query/abs.py
python3 ./test.py -f 2-query/ceil.py python3 ./test.py -f 2-query/ceil.py
python3 ./test.py -f 2-query/floor.py python3 ./test.py -f 2-query/floor.py
@ -83,6 +83,7 @@ python3 ./test.py -f 2-query/diff.py
python3 ./test.py -f 2-query/sample.py python3 ./test.py -f 2-query/sample.py
python3 ./test.py -f 2-query/function_diff.py python3 ./test.py -f 2-query/function_diff.py
python3 ./test.py -f 2-query/unique.py python3 ./test.py -f 2-query/unique.py
python3 ./test.py -f 2-query/stateduration.py
python3 ./test.py -f 7-tmq/basic5.py python3 ./test.py -f 7-tmq/basic5.py
python3 ./test.py -f 7-tmq/subscribeDb.py python3 ./test.py -f 7-tmq/subscribeDb.py

10
tests/system-test/test.py
View File

@ -37,6 +37,7 @@ if __name__ == "__main__":
masterIp = "" masterIp = ""
testCluster = False testCluster = False
valgrind = 0 valgrind = 0
killValgrind = 1
logSql = True logSql = True
stop = 0 stop = 0
restart = False restart = False
@ -45,8 +46,8 @@ if __name__ == "__main__":
windows = 1 windows = 1
updateCfgDict = {} updateCfgDict = {}
execCmd = "" execCmd = ""
opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:e:', [ opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:', [
'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'execCmd']) 'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'killv', 'execCmd'])
for key, value in opts: for key, value in opts:
if key in ['-h', '--help']: if key in ['-h', '--help']:
tdLog.printNoPrefix( tdLog.printNoPrefix(
@ -60,6 +61,7 @@ if __name__ == "__main__":
tdLog.printNoPrefix('-g valgrind Test Flag') tdLog.printNoPrefix('-g valgrind Test Flag')
tdLog.printNoPrefix('-r taosd restart test') tdLog.printNoPrefix('-r taosd restart test')
tdLog.printNoPrefix('-d update cfg dict, base64 json str') tdLog.printNoPrefix('-d update cfg dict, base64 json str')
tdLog.printNoPrefix('-k not kill valgrind processer')
tdLog.printNoPrefix('-e eval str to run') tdLog.printNoPrefix('-e eval str to run')
sys.exit(0) sys.exit(0)
@ -100,6 +102,9 @@ if __name__ == "__main__":
print('updateCfgDict convert fail.') print('updateCfgDict convert fail.')
sys.exit(0) sys.exit(0)
if key in ['-k', '--killValgrind']:
killValgrind = 0
if key in ['-e', '--execCmd']: if key in ['-e', '--execCmd']:
try: try:
execCmd = base64.b64decode(value.encode()).decode() execCmd = base64.b64decode(value.encode()).decode()
@ -189,6 +194,7 @@ if __name__ == "__main__":
else: else:
tdCases.runAllWindows(conn) tdCases.runAllWindows(conn)
else: else:
tdDnodes.setKillValgrind(killValgrind)
tdDnodes.init(deployPath, masterIp) tdDnodes.init(deployPath, masterIp)
tdDnodes.setTestCluster(testCluster) tdDnodes.setTestCluster(testCluster)
tdDnodes.setValgrind(valgrind) tdDnodes.setValgrind(valgrind)

4
tests/test/c/sdbDump.c
View File

@ -283,9 +283,7 @@ void dumpTrans(SSdb *pSdb, SJson *json) {
tjsonAddStringToObject(item, "createdTime", i642str(pObj->createdTime)); tjsonAddStringToObject(item, "createdTime", i642str(pObj->createdTime));
tjsonAddStringToObject(item, "dbUid", i642str(pObj->dbUid)); tjsonAddStringToObject(item, "dbUid", i642str(pObj->dbUid));
tjsonAddStringToObject(item, "dbname", pObj->dbname); tjsonAddStringToObject(item, "dbname", pObj->dbname);
tjsonAddIntegerToObject(item, "redoLogNum", taosArrayGetSize(pObj->redoLogs)); tjsonAddIntegerToObject(item, "commitLogNum", taosArrayGetSize(pObj->commitActions));
tjsonAddIntegerToObject(item, "undoLogNum", taosArrayGetSize(pObj->undoLogs));
tjsonAddIntegerToObject(item, "commitLogNum", taosArrayGetSize(pObj->commitLogs));
tjsonAddIntegerToObject(item, "redoActionNum", taosArrayGetSize(pObj->redoActions)); tjsonAddIntegerToObject(item, "redoActionNum", taosArrayGetSize(pObj->redoActions));
tjsonAddIntegerToObject(item, "undoActionNum", taosArrayGetSize(pObj->undoActions)); tjsonAddIntegerToObject(item, "undoActionNum", taosArrayGetSize(pObj->undoActions));