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

Merge branch '3.0' into fix/mnode

This commit is contained in:
Shengliang Guan 2022-07-20 16:16:49 +08:00
parent e9c408df6b 5a0668dadd
commit f52ad00d12
21 changed files with 439 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/client/src/clientSml.c
View File

@ -2210,7 +2210,7 @@ static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
(SSmlSTableMeta **)taosHashGet(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen);
if (tableMeta) { // update meta
ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
if (!hasTable && ret) {
if (!hasTable && ret == TSDB_CODE_SUCCESS) {
ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
}
if (ret != TSDB_CODE_SUCCESS) {

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

@ -44,6 +44,10 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM
SSyncMgmt *pMgmt = &pMnode->syncMgmt;
SSdbRaw *pRaw = pMsg->pCont;
// delete msg handle
SRpcMsg rpcMsg = {0};
syncGetAndDelRespRpc(pMnode->syncMgmt.sync, cbMeta.seqNum, &rpcMsg.info);
int32_t transId = sdbGetIdFromRaw(pMnode->pSdb, pRaw);
pMgmt->errCode = cbMeta.code;
mDebug("trans:%d, is proposed, saved:%d code:0x%x, apply index:%" PRId64 " term:%" PRIu64 " config:%" PRId64

1
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -2807,7 +2807,6 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTabl
if (pCond->suid != 0) {
(*ppReader)->pSchema = metaGetTbTSchema((*ppReader)->pTsdb->pVnode->pMeta, (*ppReader)->suid, -1);
// ASSERT((*ppReader)->pSchema);
} else if (taosArrayGetSize(pTableList) > 0) {
STableKeyInfo* pKey = taosArrayGet(pTableList, 0);
(*ppReader)->pSchema = metaGetTbTSchema((*ppReader)->pTsdb->pVnode->pMeta, pKey->uid, -1);

69
source/libs/function/src/builtinsimpl.c
View File

@ -165,6 +165,7 @@ typedef struct SElapsedInfo {
typedef struct STwaInfo {
double dOutput;
bool isNull;
SPoint1 p;
STimeWindow win;
} STwaInfo;
@ -5181,8 +5182,9 @@ bool twaFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
}
STwaInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
pInfo->p.key = INT64_MIN;
pInfo->win = TSWINDOW_INITIALIZER;
pInfo->isNull = false;
pInfo->p.key = INT64_MIN;
pInfo->win = TSWINDOW_INITIALIZER;
return true;
}
@ -5208,27 +5210,47 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
SPoint1* last = &pInfo->p;
int32_t numOfElems = 0;
if (IS_NULL_TYPE(pInputCol->info.type)) {
pInfo->isNull = true;
goto _twa_over;
}
int32_t i = pInput->startRowIndex;
if (pCtx->start.key != INT64_MIN) {
ASSERT((pCtx->start.key < tsList[i] && pCtx->order == TSDB_ORDER_ASC) ||
(pCtx->start.key > tsList[i] && pCtx->order == TSDB_ORDER_DESC));
ASSERT(last->key == INT64_MIN);
last->key = tsList[i];
for (; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
GET_TYPED_DATA(last->val, double, pInputCol->info.type, colDataGetData(pInputCol, i));
last->key = tsList[i];
pInfo->dOutput += twa_get_area(pCtx->start, *last);
pInfo->win.skey = pCtx->start.key;
numOfElems++;
i += 1;
GET_TYPED_DATA(last->val, double, pInputCol->info.type, colDataGetData(pInputCol, i));
pInfo->dOutput += twa_get_area(pCtx->start, *last);
pInfo->win.skey = pCtx->start.key;
numOfElems++;
i += 1;
break;
}
} else if (pInfo->p.key == INT64_MIN) {
last->key = tsList[i];
GET_TYPED_DATA(last->val, double, pInputCol->info.type, colDataGetData(pInputCol, i));
for (; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
pInfo->win.skey = last->key;
numOfElems++;
i += 1;
last->key = tsList[i];
GET_TYPED_DATA(last->val, double, pInputCol->info.type, colDataGetData(pInputCol, i));
pInfo->win.skey = last->key;
numOfElems++;
i += 1;
break;
}
}
SPoint1 st = {0};
@ -5241,6 +5263,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5255,6 +5278,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5268,6 +5292,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5281,6 +5306,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5294,6 +5320,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5307,6 +5334,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5320,6 +5348,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5333,6 +5362,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5346,6 +5376,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5359,6 +5390,7 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
continue;
}
numOfElems++;
INIT_INTP_POINT(st, tsList[i], val[i]);
pInfo->dOutput += twa_get_area(pInfo->p, st);
@ -5379,7 +5411,12 @@ int32_t twaFunction(SqlFunctionCtx* pCtx) {
pInfo->win.ekey = pInfo->p.key;
SET_VAL(pResInfo, numOfElems, 1);
_twa_over:
if (numOfElems == 0) {
pInfo->isNull = true;
}
SET_VAL(pResInfo, 1, 1);
return TSDB_CODE_SUCCESS;
}
@ -5400,8 +5437,8 @@ int32_t twaFinalize(struct SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
STwaInfo* pInfo = (STwaInfo*)GET_ROWCELL_INTERBUF(pResInfo);
if (pResInfo->numOfRes == 0) {
pResInfo->isNullRes = 1;
if (pInfo->isNull == true) {
pResInfo->numOfRes = 0;
} else {
if (pInfo->win.ekey == pInfo->win.skey) {
pInfo->dOutput = pInfo->p.val;

13
source/libs/sync/src/syncElection.c
View File

@ -71,6 +71,8 @@ int32_t syncNodeRequestVotePeersSnapshot(SSyncNode* pSyncNode) {
}
int32_t syncNodeElect(SSyncNode* pSyncNode) {
syncNodeEventLog(pSyncNode, "begin election");
int32_t ret = 0;
if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
syncNodeFollower2Candidate(pSyncNode);
@ -120,12 +122,15 @@ int32_t syncNodeRequestVote(SSyncNode* pSyncNode, const SRaftId* destRaftId, con
int32_t ret = 0;
do {
char host[128];
char host[64];
uint16_t port;
syncUtilU642Addr(destRaftId->addr, host, sizeof(host), &port);
sDebug("vgId:%d, send sync-request-vote to %s:%d, {term:%" PRIu64 ", last-index:%" PRId64 ", last-term:%" PRIu64
"}",
pSyncNode->vgId, host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"send sync-request-vote to %s:%d {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64 "", host, port,
pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
SRpcMsg rpcMsg;

19
source/libs/sync/src/syncMain.c
View File

@ -999,7 +999,18 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
// init TLA+ log vars
pSyncNode->pLogStore = logStoreCreate(pSyncNode);
ASSERT(pSyncNode->pLogStore != NULL);
pSyncNode->commitIndex = SYNC_INDEX_INVALID;
SyncIndex commitIndex = SYNC_INDEX_INVALID;
if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
SSnapshot snapshot = {0};
int32_t code = pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
ASSERT(code == 0);
if (snapshot.lastApplyIndex > commitIndex) {
commitIndex = snapshot.lastApplyIndex;
syncNodeEventLog(pSyncNode, "reset commit index by snapshot");
}
}
pSyncNode->commitIndex = commitIndex;
// timer ms init
pSyncNode->pingBaseLine = PING_TIMER_MS;
@ -2061,21 +2072,21 @@ void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
ASSERT(pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER);
pSyncNode->state = TAOS_SYNC_STATE_CANDIDATE;
syncNodeLog2("==state change syncNodeFollower2Candidate==", pSyncNode);
syncNodeEventLog(pSyncNode, "follower to candidate");
}
void syncNodeLeader2Follower(SSyncNode* pSyncNode) {
ASSERT(pSyncNode->state == TAOS_SYNC_STATE_LEADER);
syncNodeBecomeFollower(pSyncNode, "leader to follower");
syncNodeLog2("==state change syncNodeLeader2Follower==", pSyncNode);
syncNodeEventLog(pSyncNode, "leader to follower");
}
void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
syncNodeBecomeFollower(pSyncNode, "candidate to follower");
syncNodeLog2("==state change syncNodeCandidate2Follower==", pSyncNode);
syncNodeEventLog(pSyncNode, "candidate to follower");
}
// raft vote --------------

16
source/libs/sync/src/syncRequestVote.c
View File

@ -45,8 +45,6 @@
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
int32_t ret = 0;
syncRequestVoteLog2("==syncNodeOnRequestVoteCb==", pMsg);
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) {
do {
@ -55,8 +53,8 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote from %s:%d, term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
", maybe replica already dropped",
"recv sync-request-vote from %s:%d, {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
"}, maybe replica already dropped",
host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm);
syncNodeEventLog(ths, logBuf);
} while (0);
@ -98,8 +96,8 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote from %s:%d, term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
", reply-grant:%d",
"recv sync-request-vote from %s:%d, {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
"}, reply-grant:%d",
host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm, pReply->voteGranted);
syncNodeEventLog(ths, logBuf);
} while (0);
@ -220,8 +218,8 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) {
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote from %s:%d, term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
", maybe replica already dropped",
"recv sync-request-vote from %s:%d, {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
"}, maybe replica already dropped",
host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm);
syncNodeEventLog(ths, logBuf);
} while (0);
@ -262,7 +260,7 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) {
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote from %s:%d, {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64
", reply-grant:%d}",
"}, reply-grant:%d",
host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm, pReply->voteGranted);
syncNodeEventLog(ths, logBuf);
} while (0);

107
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -40,22 +40,41 @@
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {
int32_t ret = 0;
// print log
char logBuf[128] = {0};
snprintf(logBuf, sizeof(logBuf), "==syncNodeOnRequestVoteReplyCb== term:%" PRIu64, ths->pRaftStore->currentTerm);
syncRequestVoteReplyLog2(logBuf, pMsg);
// trace log
do {
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d} ", host,
port, pMsg->term, pMsg->voteGranted);
syncNodeEventLog(ths, logBuf);
} while (0);
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) {
sInfo("recv SyncRequestVoteReply, maybe replica already dropped");
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, maybe replica dropped", host, port,
pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
// drop stale response
if (pMsg->term < ths->pRaftStore->currentTerm) {
sTrace("recv SyncRequestVoteReply, drop stale response, receive_term:%" PRIu64 " current_term:%" PRIu64, pMsg->term,
ths->pRaftStore->currentTerm);
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, drop stale response", host, port,
pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
// ASSERT(!(pMsg->term > ths->pRaftStore->currentTerm));
@ -65,12 +84,14 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
// }
if (pMsg->term > ths->pRaftStore->currentTerm) {
char logBuf[128] = {0};
snprintf(logBuf, sizeof(logBuf), "syncNodeOnRequestVoteReplyCb error term, receive:%" PRIu64 " current:%" PRIu64,
pMsg->term, ths->pRaftStore->currentTerm);
syncNodePrint2(logBuf, ths);
sError("%s", logBuf);
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, error term",
host, port, pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
ASSERT(pMsg->term == ths->pRaftStore->currentTerm);
@ -99,7 +120,7 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
}
}
return ret;
return 0;
}
#if 0
@ -164,22 +185,41 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {
int32_t ret = 0;
// print log
char logBuf[128] = {0};
snprintf(logBuf, sizeof(logBuf), "recv SyncRequestVoteReply, term:%" PRIu64, ths->pRaftStore->currentTerm);
syncRequestVoteReplyLog2(logBuf, pMsg);
// trace log
do {
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d} ", host,
port, pMsg->term, pMsg->voteGranted);
syncNodeEventLog(ths, logBuf);
} while (0);
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) {
sInfo("recv SyncRequestVoteReply, maybe replica already dropped");
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, maybe replica dropped", host, port,
pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
// drop stale response
if (pMsg->term < ths->pRaftStore->currentTerm) {
sTrace("recv SyncRequestVoteReply, drop stale response, receive_term:%" PRIu64 " current_term:%" PRIu64, pMsg->term,
ths->pRaftStore->currentTerm);
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf),
"recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, drop stale response", host, port,
pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
// ASSERT(!(pMsg->term > ths->pRaftStore->currentTerm));
@ -189,13 +229,14 @@ int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteRepl
// }
if (pMsg->term > ths->pRaftStore->currentTerm) {
char logBuf[128] = {0};
snprintf(logBuf, sizeof(logBuf),
"recv SyncRequestVoteReply, error term, receive_term:%" PRIu64 " current_term:%" PRIu64, pMsg->term,
ths->pRaftStore->currentTerm);
syncNodePrint2(logBuf, ths);
sError("%s", logBuf);
return ret;
char host[64];
uint16_t port;
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, error term",
host, port, pMsg->term, pMsg->voteGranted);
syncNodeErrorLog(ths, logBuf);
return -1;
}
ASSERT(pMsg->term == ths->pRaftStore->currentTerm);
@ -224,5 +265,5 @@ int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteRepl
}
}
return ret;
return 0;
}

6
source/libs/sync/src/syncSnapshot.c
View File

@ -573,6 +573,12 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
}
// maybe update term
if (pReceiver->snapshot.lastApplyTerm > pReceiver->pSyncNode->pRaftStore->currentTerm) {
pReceiver->pSyncNode->pRaftStore->currentTerm = pReceiver->snapshot.lastApplyTerm;
raftStorePersist(pReceiver->pSyncNode->pRaftStore);
}
// stop writer, apply data
code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true,
&(pReceiver->snapshot));

2
source/libs/transport/inc/transComm.h
View File

@ -392,7 +392,7 @@ typedef struct SDelayQueue {
} SDelayQueue;
int transDQCreate(uv_loop_t* loop, SDelayQueue** queue);
void transDQDestroy(SDelayQueue* queue);
void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg));
int transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
bool transEpSetIsEqual(SEpSet* a, SEpSet* b);

BIN
source/libs/transport/src/.transComm.c.swo Normal file
View File

Binary file not shown.

10
source/libs/transport/src/transCli.c
View File

@ -140,7 +140,7 @@ static void destroyUserdata(STransMsg* userdata);
static int cliRBChoseIdx(STrans* pTransInst);
static void destroyCmsg(SCliMsg* cmsg);
static void destroyCmsg(void* cmsg);
static void transDestroyConnCtx(STransConnCtx* ctx);
// thread obj
static SCliThrd* createThrdObj();
@ -198,6 +198,7 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
} \
destroyCmsg(pMsg); \
cliReleaseUnfinishedMsg(conn); \
transQueueClear(&conn->cliMsgs); \
addConnToPool(((SCliThrd*)conn->hostThrd)->pool, conn); \
return; \
} \
@ -545,6 +546,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
STrans* pTransInst = thrd->pTransInst;
conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
cliReleaseUnfinishedMsg(conn);
transQueueClear(&conn->cliMsgs);
transCtxCleanup(&conn->ctx);
conn->status = ConnInPool;
@ -645,6 +647,7 @@ static void cliDestroy(uv_handle_t* handle) {
conn->stream->data = NULL;
taosMemoryFree(conn->stream);
transCtxCleanup(&conn->ctx);
cliReleaseUnfinishedMsg(conn);
transQueueDestroy(&conn->cliMsgs);
tTrace("%s conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
transReqQueueClear(&conn->wreqQueue);
@ -962,7 +965,8 @@ static void destroyUserdata(STransMsg* userdata) {
transFreeMsg(userdata->pCont);
userdata->pCont = NULL;
}
static void destroyCmsg(SCliMsg* pMsg) {
static void destroyCmsg(void* arg) {
SCliMsg* pMsg = arg;
if (pMsg == NULL) {
return;
}
@ -1001,7 +1005,7 @@ static void destroyThrdObj(SCliThrd* pThrd) {
TRANS_DESTROY_ASYNC_POOL_MSG(pThrd->asyncPool, SCliMsg, destroyCmsg);
transDestroyAsyncPool(pThrd->asyncPool);
transDQDestroy(pThrd->delayQueue);
transDQDestroy(pThrd->delayQueue, destroyCmsg);
taosMemoryFree(pThrd->loop);
taosMemoryFree(pThrd);
}

7
source/libs/transport/src/transComm.c
View File

@ -456,7 +456,7 @@ int transDQCreate(uv_loop_t* loop, SDelayQueue** queue) {
return 0;
}
void transDQDestroy(SDelayQueue* queue) {
void transDQDestroy(SDelayQueue* queue, void (*freeFunc)(void* arg)) {
taosMemoryFree(queue->timer);
while (heapSize(queue->heap) > 0) {
@ -467,6 +467,11 @@ void transDQDestroy(SDelayQueue* queue) {
heapRemove(queue->heap, minNode);
SDelayTask* task = container_of(minNode, SDelayTask, node);
STaskArg* arg = task->arg;
freeFunc(arg->param1);
taosMemoryFree(arg);
taosMemoryFree(task);
}
heapDestroy(queue->heap);

2
tests/script/tsim/valgrind/checkError3.sim
View File

@ -90,7 +90,7 @@ $null=
system_content sh/checkValgrind.sh -n dnode1
print cmd return result ----> [ $system_content ]
if $system_content > 0 then
if $system_content > 2 then
return -1
endi

12
tests/system-test/1-insert/alter_table.py
View File

@ -211,10 +211,10 @@ class TDTestCase:
for error in [constant.INT_UN_MIN-1,constant.INT_UN_MAX+1]:
tdSql.error(f'alter table {self.stbname}_{i} set tag {k} = {error}')
#! bug TD-17106
# elif v.lower() == 'bigint unsigned':
# self.tag_check(i,k,tag_unbigint)
# for error in [constant.BIGINT_UN_MIN-1,constant.BIGINT_UN_MAX+1]:
# tdSql.error(f'alter table {self.stbname}_{i} set tag {k} = {error}')
elif v.lower() == 'bigint unsigned':
self.tag_check(i,k,tag_unbigint)
for error in [constant.BIGINT_UN_MIN-1,constant.BIGINT_UN_MAX+1]:
tdSql.error(f'alter table {self.stbname}_{i} set tag {k} = {error}')
elif v.lower() == 'bool':
self.tag_check(i,k,tag_bool)
elif v.lower() == 'float':
@ -225,8 +225,8 @@ class TDTestCase:
else:
tdLog.exit(f'select {k} from {self.stbname}_{i},data check failure')
#! bug TD-17106
# for error in [constant.FLOAT_MIN*1.1,constant.FLOAT_MAX*1.1]:
# tdSql.error(f'alter table {self.stbname}_{i} set tag {k} = {error}')
for error in [constant.FLOAT_MIN*1.1,constant.FLOAT_MAX*1.1]:
tdSql.error(f'alter table {self.stbname}_{i} set tag {k} = {error}')
elif v.lower() == 'double':
tdSql.execute(f'alter table {self.stbname}_{i} set tag {k} = {tag_double}')
tdSql.query(f'select {k} from {self.stbname}_{i}')

159
tests/system-test/1-insert/delete_data.py
View File

@ -25,12 +25,13 @@ from util.sqlset import TDSetSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(),logSql)
tdSql.init(conn.cursor())
self.dbname = 'db_test'
self.setsql = TDSetSql()
self.stbname = 'stb'
self.ntbname = 'ntb'
self.rowNum = 10
self.tbnum = 20
self.rowNum = 5
self.tbnum = 2
self.ts = 1537146000000
self.binary_str = 'taosdata'
self.nchar_str = '涛思数据'
@ -51,6 +52,7 @@ class TDTestCase:
'col13': f'nchar({self.str_length})',
}
self.tinyint_val = random.randint(constant.TINYINT_MIN,constant.TINYINT_MAX)
self.smallint_val = random.randint(constant.SMALLINT_MIN,constant.SMALLINT_MAX)
self.int_val = random.randint(constant.INT_MIN,constant.INT_MAX)
@ -107,32 +109,50 @@ class TDTestCase:
tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['binary']}")''')
elif 'nchar' in col_type.lower():
tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['nchar']}")''')
def delete_all_data(self,tbname,col_type,row_num,base_data,dbname):
def delete_all_data(self,tbname,col_type,row_num,base_data,dbname,tb_type,tb_num=1):
tdSql.execute(f'delete from {tbname}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select * from {tbname}')
tdSql.checkRows(0)
self.insert_base_data(col_type,tbname,row_num,base_data)
if tb_type == 'ntb' or tb_type == 'ctb':
self.insert_base_data(col_type,tbname,row_num,base_data)
elif tb_type == 'stb':
for i in range(tb_num):
self.insert_base_data(col_type,f'{tbname}_{i}',row_num,base_data)
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select * from {tbname}')
tdSql.checkRows(row_num)
def delete_one_row(self,tbname,column_type,column_name,base_data,dbname):
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(row_num)
elif tb_type =='stb':
tdSql.checkRows(row_num*tb_num)
def delete_one_row(self,tbname,column_type,column_name,base_data,row_num,dbname,tb_type,tb_num=1):
tdSql.execute(f'delete from {tbname} where ts={self.ts}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {column_name} from {tbname}')
tdSql.checkRows(self.rowNum-1)
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(row_num-1)
elif tb_type == 'stb':
tdSql.checkRows((row_num-1)*tb_num)
tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}')
tdSql.checkRows(0)
if 'binary' in column_type.lower():
tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['binary']}")''')
elif 'nchar' in column_type.lower():
tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['nchar']}")''')
else:
tdSql.execute(f'insert into {tbname} values({self.ts},{base_data[column_type]})')
if tb_type == 'ntb' or tb_type == 'ctb':
if 'binary' in column_type.lower():
tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['binary']}")''')
elif 'nchar' in column_type.lower():
tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['nchar']}")''')
else:
tdSql.execute(f'insert into {tbname} values({self.ts},{base_data[column_type]})')
elif tb_type == 'stb':
for i in range(tb_num):
if 'binary' in column_type.lower():
tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['binary']}")''')
elif 'nchar' in column_type.lower():
tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['nchar']}")''')
else:
tdSql.execute(f'insert into {tbname}_{i} values({self.ts},{base_data[column_type]})')
tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}')
if column_type.lower() == 'float' or column_type.lower() == 'double':
if abs(tdSql.queryResult[0][0] - base_data[column_type]) / base_data[column_type] <= 0.0001:
@ -145,11 +165,55 @@ class TDTestCase:
tdSql.checkEqual(tdSql.queryResult[0][0],base_data['nchar'])
else:
tdSql.checkEqual(tdSql.queryResult[0][0],base_data[column_type])
def delete_rows(self):
pass
def delete_rows(self,dbname,tbname,col_name,col_type,base_data,row_num,tb_type,tb_num=1):
for i in range(row_num):
tdSql.execute(f'delete from {tbname} where ts>{self.ts+i}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {col_name} from {tbname}')
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(i+1)
self.insert_base_data(col_type,tbname,row_num,base_data)
elif tb_type == 'stb':
tdSql.checkRows((i+1)*tb_num)
for j in range(tb_num):
self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data)
for i in range(row_num):
tdSql.execute(f'delete from {tbname} where ts>={self.ts+i}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {col_name} from {tbname}')
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(i)
self.insert_base_data(col_type,tbname,row_num,base_data)
elif tb_type == 'stb':
tdSql.checkRows(i*tb_num)
for j in range(tb_num):
self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data)
for i in range(row_num):
tdSql.execute(f'delete from {tbname} where ts<={self.ts+i}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {col_name} from {tbname}')
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(row_num-i-1)
self.insert_base_data(col_type,tbname,row_num,base_data)
elif tb_type == 'stb':
tdSql.checkRows((row_num-i-1)*tb_num)
for j in range(tb_num):
self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data)
for i in range(row_num):
tdSql.execute(f'delete from {tbname} where ts<{self.ts+i}')
tdSql.execute(f'flush database {dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {col_name} from {tbname}')
if tb_type == 'ntb' or tb_type == 'ctb':
tdSql.checkRows(row_num-i)
self.insert_base_data(col_type,tbname,row_num,base_data)
elif tb_type == 'stb':
tdSql.checkRows((row_num-i)*tb_num)
for j in range(tb_num):
self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data)
def delete_error(self,tbname,column_name,column_type,base_data):
for error_list in ['',f'ts = {self.ts} and',f'ts = {self.ts} or']:
if 'binary' in column_type.lower():
@ -157,31 +221,56 @@ class TDTestCase:
elif 'nchar' in column_type.lower():
tdSql.error(f'''delete from {tbname} where {error_list} {column_name} ="{base_data['nchar']}"''')
else:
tdSql.error('delete from {tbname} where {error_list} {column_name} = {base_data[column_type]}')
tdSql.error(f'delete from {tbname} where {error_list} {column_name} = {base_data[column_type]}')
def delete_data_ntb(self):
tdSql.execute(f'create database if not exists {self.dbname}')
tdSql.execute(f'use {self.dbname}')
for col_name,col_type in self.column_dict.items():
tdSql.execute(f'create table {self.ntbname} (ts timestamp,{col_name} {col_type})')
self.insert_base_data(col_type,self.ntbname,self.rowNum,self.base_data)
self.delete_one_row(self.ntbname,col_type,col_name,self.base_data,self.dbname)
self.delete_all_data(self.ntbname,col_type,self.rowNum,self.base_data,self.dbname)
self.delete_one_row(self.ntbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'ntb')
self.delete_all_data(self.ntbname,col_type,self.rowNum,self.base_data,self.dbname,'ntb')
self.delete_error(self.ntbname,col_name,col_type,self.base_data)
for i in range(self.rowNum):
tdSql.execute(f'delete from {self.ntbname} where ts>{self.ts+i}')
tdSql.execute(f'flush database {self.dbname}')
tdSql.execute('reset query cache')
tdSql.query(f'select {col_name} from {self.ntbname}')
tdSql.checkRows(i+1)
self.insert_base_data(col_type,self.ntbname,self.rowNum,self.base_data)
self.delete_rows(self.dbname,self.ntbname,col_name,col_type,self.base_data,self.rowNum,'ntb')
for func in ['first','last']:
tdSql.query(f'select {func}(*) from {self.ntbname}')
tdSql.execute(f'drop table {self.ntbname}')
tdSql.execute(f'drop database {self.dbname}')
def delete_data_ctb(self):
tdSql.execute(f'create database if not exists {self.dbname}')
tdSql.execute(f'use {self.dbname}')
for col_name,col_type in self.column_dict.items():
tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)')
for i in range(self.tbnum):
tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)')
self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data)
self.delete_one_row(f'{self.stbname}_{i}',col_type,col_name,self.base_data,self.rowNum,self.dbname,'ctb')
self.delete_all_data(f'{self.stbname}_{i}',col_type,self.rowNum,self.base_data,self.dbname,'ctb')
self.delete_error(f'{self.stbname}_{i}',col_name,col_type,self.base_data)
self.delete_rows(self.dbname,f'{self.stbname}_{i}',col_name,col_type,self.base_data,self.rowNum,'ctb')
for func in ['first','last']:
tdSql.query(f'select {func}(*) from {self.stbname}_{i}')
tdSql.execute(f'drop table {self.stbname}')
def delete_data_stb(self):
tdSql.execute(f'create database if not exists {self.dbname}')
tdSql.execute(f'use {self.dbname}')
for col_name,col_type in self.column_dict.items():
tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)')
for i in range(self.tbnum):
tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)')
self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data)
self.delete_error(self.stbname,col_name,col_type,self.base_data)
self.delete_one_row(self.stbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'stb',self.tbnum)
self.delete_all_data(self.stbname,col_type,self.rowNum,self.base_data,self.dbname,'stb',self.tbnum)
self.delete_rows(self.dbname,self.stbname,col_name,col_type,self.base_data,self.rowNum,'stb',self.tbnum)
for func in ['first','last']:
tdSql.query(f'select {func}(*) from {self.stbname}')
tdSql.execute(f'drop table {self.stbname}')
tdSql.execute(f'drop database {self.dbname}')
def run(self):
self.delete_data_ntb()
self.delete_data_ctb()
self.delete_data_stb()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)

2
tests/system-test/2-query/max_partition.py
View File

@ -184,7 +184,7 @@ class TDTestCase:
tdSql.query("select c1 , twa(c1) from stb partition by c1 order by c1")
tdSql.checkRows(11)
tdSql.checkData(0,1,0.000000000)
tdSql.checkData(0,1,None)
tdSql.query("select c1 , irate(c1) from stb partition by c1 order by c1")
tdSql.checkRows(11)

38
tests/system-test/7-tmq/tmqAutoCreateTbl.py
View File

@ -16,6 +16,7 @@ from tmqCommon import *
class TDTestCase:
def __init__(self):
self.snapshot = 0
self.vgroups = 4
self.ctbNum = 1000
self.rowsPerTbl = 1000
@ -44,7 +45,7 @@ class TDTestCase:
'pollDelay': 3,
'showMsg': 1,
'showRow': 1,
'snapshot': 1}
'snapshot': 0}
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
@ -84,13 +85,14 @@ class TDTestCase:
'ctbStartIdx': 0,
'ctbNum': 1000,
'rowsPerTbl': 1000,
'batchNum': 400,
'batchNum': 1000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5,
'showMsg': 1,
'showRow': 1,
'snapshot': 1}
'snapshot': 0}
paraDict['snapshot'] = self.snapshot
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
@ -131,10 +133,10 @@ class TDTestCase:
totalConsumeRows += resultList[i]
tdSql.query(queryString)
totalRowsInserted = tdSql.getRows()
totalRowsFromQuery = tdSql.getRows()
if totalConsumeRows != totalRowsInserted:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsInserted))
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsFromQuery))
if totalConsumeRows != totalRowsFromQuery:
tdLog.exit("tmq consume rows error!")
tdSql.query("drop topic %s"%topicFromStb1)
@ -163,6 +165,7 @@ class TDTestCase:
'showRow': 1,
'snapshot': 0}
paraDict['snapshot'] = self.snapshot
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
@ -180,12 +183,13 @@ class TDTestCase:
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
# queryString = "select ts, c1, c2 from %s.%s "%(paraDict['dbName'], paraDict['stbName'])
queryString = "select ts, c1, c2 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName'])
sqlString = "create topic %s as %s" %(topicFromStb1, queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
consumerId = 0
consumerId = 1
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2
topicList = topicFromStb1
ifcheckdata = 0
@ -210,10 +214,10 @@ class TDTestCase:
totalConsumeRows += resultList[i]
tdSql.query(queryString)
totalRowsInserted = tdSql.getRows()
totalRowsFromQuery = tdSql.getRows()
if totalConsumeRows != totalRowsInserted:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsInserted))
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsFromQuery))
if totalConsumeRows != totalRowsFromQuery:
tdLog.exit("tmq consume rows error!")
tdSql.query("drop topic %s"%topicFromStb1)
@ -222,10 +226,18 @@ class TDTestCase:
def run(self):
tdSql.prepare()
self.prepareTestEnv()
tdLog.printNoPrefix("=============================================")
tdLog.printNoPrefix("======== snapshot is 0: only consume from wal")
self.tmqCase1()
# self.tmqCase2() # TD-17267
self.tmqCase2()
self.prepareTestEnv()
tdLog.printNoPrefix("====================================================================")
tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal")
self.snapshot = 1
self.tmqCase1()
self.tmqCase2()
def stop(self):

112
tests/system-test/7-tmq/tmqDnodeRestart.py
View File

@ -16,6 +16,7 @@ from tmqCommon import *
class TDTestCase:
def __init__(self):
self.snapshot = 0
self.vgroups = 2
self.ctbNum = 100
self.rowsPerTbl = 10000
@ -37,15 +38,16 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb',
'ctbStartIdx': 0,
'ctbNum': 500,
'rowsPerTbl': 1000,
'batchNum': 500,
'ctbNum': 100,
'rowsPerTbl': 10000,
'batchNum': 100,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 3,
'showMsg': 1,
'showRow': 1,
'snapshot': 0}
paraDict['snapshot'] = self.snapshot
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
@ -81,30 +83,31 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb',
'ctbStartIdx': 0,
'ctbNum': 1000,
'rowsPerTbl': 1000,
'batchNum': 400,
'ctbNum': 100,
'rowsPerTbl': 10000,
'batchNum': 100,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5,
'showMsg': 1,
'showRow': 1,
'snapshot': 1}
'snapshot': 0}
# paraDict['vgroups'] = self.vgroups
# paraDict['ctbNum'] = self.ctbNum
# paraDict['rowsPerTbl'] = self.rowsPerTbl
paraDict['snapshot'] = self.snapshot
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("insert data")
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
# tmqCom.initConsumerTable()
# tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
# tdLog.info("create stb")
# tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
# tdLog.info("create ctb")
# tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
# ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
# tdLog.info("insert data")
# tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
# ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
@ -132,7 +135,7 @@ class TDTestCase:
tdLog.info("================= restart dnode ===========================")
tdDnodes.stop(1)
tdDnodes.start(1)
time.sleep(5)
time.sleep(3)
tdLog.info("insert process end, and start to check consume result")
expectRows = 1
@ -142,10 +145,10 @@ class TDTestCase:
totalConsumeRows += resultList[i]
tdSql.query(queryString)
totalRowsInserted = tdSql.getRows()
totalRowsFromQury = tdSql.getRows()
if totalConsumeRows != totalRowsInserted:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsInserted))
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsFromQury))
if totalConsumeRows != totalRowsFromQury:
tdLog.exit("tmq consume rows error!")
tdSql.query("drop topic %s"%topicFromStb1)
@ -165,30 +168,31 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb',
'ctbStartIdx': 0,
'ctbNum': 1000,
'rowsPerTbl': 1000,
'batchNum': 1000,
'ctbNum': 100,
'rowsPerTbl': 10000,
'batchNum': 3000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5,
'showMsg': 1,
'showRow': 1,
'snapshot': 1}
'snapshot': 0}
# paraDict['vgroups'] = self.vgroups
# paraDict['ctbNum'] = self.ctbNum
# paraDict['rowsPerTbl'] = self.rowsPerTbl
paraDict['snapshot'] = self.snapshot
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("insert data")
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
# tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
# tdLog.info("create stb")
# tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
# tdLog.info("create ctb")
# tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
# ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
# tdLog.info("insert data")
# tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
# ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
@ -196,28 +200,28 @@ class TDTestCase:
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
consumerId = 0
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2
consumerId = 1
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2 + 100000
topicList = topicFromStb1
ifcheckdata = 0
ifManualCommit = 0
keyList = 'group.id:cgrp1,\
enable.auto.commit:true,\
auto.commit.interval.ms:1000,\
auto.commit.interval.ms:3000,\
auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
tdLog.info("create some new child table and insert data ")
tmqCom.insert_data_with_autoCreateTbl(tdSql,paraDict["dbName"],paraDict["stbName"],"ctb",paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"])
tmqCom.getStartCommitNotifyFromTmqsim()
tdLog.info("================= restart dnode ===========================")
tdDnodes.stop(1)
tdDnodes.start(1)
time.sleep(5)
time.sleep(3)
tdLog.info("create some new child table and insert data ")
tmqCom.insert_data_with_autoCreateTbl(tdSql,paraDict["dbName"],paraDict["stbName"],"ctb",paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"])
tdLog.info("insert process end, and start to check consume result")
expectRows = 1
@ -227,10 +231,10 @@ class TDTestCase:
totalConsumeRows += resultList[i]
tdSql.query(queryString)
totalRowsInserted = tdSql.getRows()
totalRowsFromQuery = tdSql.getRows()
if totalConsumeRows != totalRowsInserted:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsInserted))
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, totalRowsFromQuery))
if totalConsumeRows != totalRowsFromQuery:
tdLog.exit("tmq consume rows error!")
tdSql.query("drop topic %s"%topicFromStb1)
@ -239,8 +243,8 @@ class TDTestCase:
def run(self):
tdSql.prepare()
self.tmqCase1()
self.prepareTestEnv()
# self.tmqCase1()
self.tmqCase2()
def stop(self):

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

@ -33,7 +33,7 @@ python3 ./test.py -f 1-insert/create_retentions.py
python3 ./test.py -f 1-insert/table_param_ttl.py
python3 ./test.py -f 1-insert/update_data.py
python3 ./test.py -f 1-insert/delete_data.py
python3 ./test.py -f 2-query/db.py
python3 ./test.py -f 2-query/between.py
@ -184,7 +184,7 @@ python3 ./test.py -f 7-tmq/tmqConsFromTsdb-mutilVg-mutilCtb.py
python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-1ctb-funcNFilter.py
python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py
python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb.py
python3 ./test.py -f 7-tmq/tmqAutoCreateTbl.py
#python3 ./test.py -f 7-tmq/tmqAutoCreateTbl.py
#python3 ./test.py -f 7-tmq/tmqDnodeRestart.py
python3 ./test.py -f 7-tmq/tmqUpdate-1ctb.py
python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py