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 szhou/tms-wide-column

This commit is contained in:
shenglian zhou 2024-01-02 10:41:31 +08:00
parent 81b95f2135 59642c674d
commit 91a705f731
54 changed files with 1297 additions and 2404 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/tmsg.h
View File

@ -2544,12 +2544,6 @@ _err:
return NULL;
}
// this message is sent from mnode to mnode(read thread to write thread),
// so there is no need for serialization or deserialization
typedef struct {
SHashObj* rebSubHash; // SHashObj<key, SMqRebSubscribe>
} SMqDoRebalanceMsg;
typedef struct {
int64_t streamId;
int64_t checkpointId;

12
source/client/src/clientTmq.c
View File

@ -1028,11 +1028,16 @@ static void tmqMgmtInit(void) {
}
}
#define SET_ERROR_MSG(MSG) if(errstr!=NULL)snprintf(errstr,errstrLen,MSG);
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if(conf == NULL) return NULL;
if(conf == NULL) {
SET_ERROR_MSG("configure is null")
return NULL;
}
taosThreadOnce(&tmqInit, tmqMgmtInit);
if (tmqInitRes != 0) {
terrno = tmqInitRes;
SET_ERROR_MSG("tmq timer init error")
return NULL;
}
@ -1040,6 +1045,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (pTmq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tscError("failed to create consumer, groupId:%s, code:%s", conf->groupId, terrstr());
SET_ERROR_MSG("malloc tmq failed")
return NULL;
}
@ -1055,6 +1061,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
conf->groupId[0] == 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG("malloc tmq element failed or group is empty")
goto _failed;
}
@ -1086,6 +1093,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (tsem_init(&pTmq->rspSem, 0, 0) != 0) {
tscError("consumer:0x %" PRIx64 " setup failed since %s, consumer group %s", pTmq->consumerId, terrstr(),
pTmq->groupId);
SET_ERROR_MSG("init t_sem failed")
goto _failed;
}
@ -1094,11 +1102,13 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (pTmq->pTscObj == NULL) {
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
tsem_destroy(&pTmq->rspSem);
SET_ERROR_MSG("init tscObj failed")
goto _failed;
}
pTmq->refId = taosAddRef(tmqMgmt.rsetId, pTmq);
if (pTmq->refId < 0) {
SET_ERROR_MSG("add tscObj ref failed")
goto _failed;
}

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

@ -789,7 +789,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "s3PageCacheSize", tsS3PageCacheSize, 4, 1024 * 1024 * 1024, CFG_SCOPE_SERVER,
CFG_DYN_ENT_SERVER) != 0)
return -1;
if (cfgAddInt32(pCfg, "s3UploadDelaySec", tsS3UploadDelaySec, 60 * 10, 60 * 60 * 24 * 30, CFG_SCOPE_SERVER,
if (cfgAddInt32(pCfg, "s3UploadDelaySec", tsS3UploadDelaySec, 60 * 1, 60 * 60 * 24 * 30, CFG_SCOPE_SERVER,
CFG_DYN_ENT_SERVER) != 0)
return -1;

5
source/dnode/mnode/impl/inc/mndConsumer.h
View File

@ -46,10 +46,7 @@ SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw);
int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer);
int32_t mndSetConsumerDropLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer);
bool mndRebTryStart();
void mndRebEnd();
void mndRebCntInc();
void mndRebCntDec();
const char *mndConsumerStatusName(int status);
#ifdef __cplusplus
}

9
source/dnode/mnode/impl/inc/mndSubscribe.h
View File

@ -32,14 +32,13 @@ void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub);
int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName);
//static FORCE_INLINE int32_t mndMakePartitionKey(char *key, const char *cgroup, const char *topicName, int32_t vgId) {
// return snprintf(key, TSDB_PARTITION_KEY_LEN, "%d:%s:%s", vgId, cgroup, topicName);
//}
//int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topic);
int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub);
bool mndRebTryStart();
void mndRebCntInc();
void mndRebCntDec();
#ifdef __cplusplus
}
#endif

197
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -29,12 +29,6 @@
#define MND_CONSUMER_RESERVE_SIZE 64
#define MND_MAX_GROUP_PER_TOPIC 100
#define MND_CONSUMER_LOST_HB_CNT 6
#define MND_CONSUMER_LOST_CLEAR_THRESHOLD 43200
static int32_t mqRebInExecCnt = 0;
static const char *mndConsumerStatusName(int status);
static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);
static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);
@ -45,7 +39,6 @@ static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);
static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);
static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);
static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg);
static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg);
static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg);
@ -63,7 +56,7 @@ int32_t mndInitConsumer(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_SUBSCRIBE, mndProcessSubscribeReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg);
// mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg);
@ -95,36 +88,6 @@ void mndDropConsumerFromSdb(SMnode *pMnode, int64_t consumerId, SRpcHandleInfo*
return;
}
bool mndRebTryStart() {
int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1);
mDebug("tq timer, rebalance counter old val:%d", old);
return old == 0;
}
void mndRebEnd() { mndRebCntDec(); }
void mndRebCntInc() {
int32_t val = atomic_add_fetch_32(&mqRebInExecCnt, 1);
mInfo("rebalance trans start, rebalance counter:%d", val);
}
void mndRebCntDec() {
while (1) {
int32_t val = atomic_load_32(&mqRebInExecCnt);
if (val <= 0) {
mError("rebalance trans end, rebalance counter:%d should not be less equalled than 0, ignore counter desc", val);
break;
}
int32_t newVal = val - 1;
int32_t oldVal = atomic_val_compare_exchange_32(&mqRebInExecCnt, val, newVal);
if (oldVal == val) {
mDebug("rebalance trans end, rebalance counter:%d", newVal);
break;
}
}
}
static int32_t validateTopics(STrans *pTrans, const SArray *pTopicList, SMnode *pMnode, const char *pUser, bool enableReplay) {
SMqTopicObj *pTopic = NULL;
int32_t code = 0;
@ -257,162 +220,6 @@ FAIL:
return -1;
}
static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
if (pRebInfo == NULL) {
pRebInfo = tNewSMqRebSubscribe(key);
if (pRebInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));
taosMemoryFree(pRebInfo);
pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
}
return pRebInfo;
}
static void freeRebalanceItem(void *param) {
SMqRebInfo *pInfo = param;
taosArrayDestroy(pInfo->newConsumers);
taosArrayDestroy(pInfo->removedConsumers);
}
static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
SMnode *pMnode = pMsg->info.node;
SSdb *pSdb = pMnode->pSdb;
SMqConsumerObj *pConsumer;
void *pIter = NULL;
mDebug("start to process mq timer");
// rebalance cannot be parallel
if (!mndRebTryStart()) {
mInfo("mq rebalance already in progress, do nothing");
return 0;
}
SMqDoRebalanceMsg *pRebMsg = rpcMallocCont(sizeof(SMqDoRebalanceMsg));
if (pRebMsg == NULL) {
mError("failed to create the rebalance msg, size:%d, quit mq timer", (int32_t)sizeof(SMqDoRebalanceMsg));
mndRebEnd();
return TSDB_CODE_OUT_OF_MEMORY;
}
pRebMsg->rebSubHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK);
if (pRebMsg->rebSubHash == NULL) {
mError("failed to create rebalance hashmap");
rpcFreeCont(pRebMsg);
mndRebEnd();
return TSDB_CODE_OUT_OF_MEMORY;
}
taosHashSetFreeFp(pRebMsg->rebSubHash, freeRebalanceItem);
// iterate all consumers, find all modification
while (1) {
pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
if (pIter == NULL) {
break;
}
int32_t hbStatus = atomic_add_fetch_32(&pConsumer->hbStatus, 1);
int32_t status = atomic_load_32(&pConsumer->status);
mInfo("check for consumer:0x%" PRIx64 " status:%d(%s), sub-time:%" PRId64 ", createTime:%" PRId64 ", hbstatus:%d",
pConsumer->consumerId, status, mndConsumerStatusName(status), pConsumer->subscribeTime, pConsumer->createTime,
hbStatus);
if (status == MQ_CONSUMER_STATUS_READY) {
if (taosArrayGetSize(pConsumer->assignedTopics) == 0) { // unsubscribe or close
mndDropConsumerFromSdb(pMnode, pConsumer->consumerId, &pMsg->info);
} else if (hbStatus > MND_CONSUMER_LOST_HB_CNT) {
taosRLockLatch(&pConsumer->lock);
int32_t topicNum = taosArrayGetSize(pConsumer->currentTopics);
for (int32_t i = 0; i < topicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *removedTopic = taosArrayGetP(pConsumer->currentTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
}
taosRUnLockLatch(&pConsumer->lock);
}else{
int32_t newTopicNum = taosArrayGetSize(pConsumer->currentTopics);
for (int32_t i = 0; i < newTopicNum; i++) {
char * topic = taosArrayGetP(pConsumer->currentTopics, i);
SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic);
if (pSub == NULL) {
continue;
}
taosRLockLatch(&pSub->lock);
// 2.2 iterate all vg assigned to the consumer of that topic
SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));
int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);
for (int32_t j = 0; j < vgNum; j++) {
SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
SVgObj * pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId);
if (!pVgroup) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
mndMakeSubscribeKey(key, pConsumer->cgroup, topic);
mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
mInfo("vnode splitted, vgId:%d rebalance will be triggered", pVgEp->vgId);
}
mndReleaseVgroup(pMnode, pVgroup);
}
taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub);
}
}
} else if (status == MQ_CONSUMER_STATUS_LOST) {
if (hbStatus > MND_CONSUMER_LOST_CLEAR_THRESHOLD) { // clear consumer if lost a day
mndDropConsumerFromSdb(pMnode, pConsumer->consumerId, &pMsg->info);
}
} else { // MQ_CONSUMER_STATUS_REBALANCE
taosRLockLatch(&pConsumer->lock);
int32_t newTopicNum = taosArrayGetSize(pConsumer->rebNewTopics);
for (int32_t i = 0; i < newTopicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *newTopic = taosArrayGetP(pConsumer->rebNewTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, newTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
taosArrayPush(pRebSub->newConsumers, &pConsumer->consumerId);
}
int32_t removedTopicNum = taosArrayGetSize(pConsumer->rebRemovedTopics);
for (int32_t i = 0; i < removedTopicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *removedTopic = taosArrayGetP(pConsumer->rebRemovedTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
}
taosRUnLockLatch(&pConsumer->lock);
}
mndReleaseConsumer(pMnode, pConsumer);
}
if (taosHashGetSize(pRebMsg->rebSubHash) != 0) {
mInfo("mq rebalance will be triggered");
SRpcMsg rpcMsg = {
.msgType = TDMT_MND_TMQ_DO_REBALANCE,
.pCont = pRebMsg,
.contLen = sizeof(SMqDoRebalanceMsg),
};
tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
} else {
taosHashCleanup(pRebMsg->rebSubHash);
rpcFreeCont(pRebMsg);
mDebug("mq timer finished, no need to re-balance");
mndRebEnd();
}
return 0;
}
static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
int32_t code = 0;
SMnode *pMnode = pMsg->info.node;
@ -1251,7 +1058,7 @@ static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
sdbCancelFetch(pSdb, pIter);
}
static const char *mndConsumerStatusName(int status) {
const char *mndConsumerStatusName(int status) {
switch (status) {
case MQ_CONSUMER_STATUS_READY:
return "ready";

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

@ -792,7 +792,7 @@ static int32_t checkForNumOfStreams(SMnode *pMnode, SStreamObj *pStreamObj) { /
if (pStream->targetStbUid == pStreamObj->targetStbUid) {
mError("Cannot write the same stable as other stream:%s", pStream->name);
sdbCancelFetch(pMnode->pSdb, pIter);
terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
terrno = TSDB_CODE_MND_INVALID_TARGET_TABLE;
return terrno;
}
}

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

@ -27,7 +27,10 @@
#define MND_SUBSCRIBE_VER_NUMBER 2
#define MND_SUBSCRIBE_RESERVE_SIZE 64
#define MND_SUBSCRIBE_REBALANCE_CNT 3
#define MND_CONSUMER_LOST_HB_CNT 6
#define MND_CONSUMER_LOST_CLEAR_THRESHOLD 43200
static int32_t mqRebInExecCnt = 0;
static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *);
static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw);
@ -38,14 +41,7 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg);
static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg);
static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter);
static int32_t mndSetSubRedoLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
SSdbRaw *pRedoRaw = mndSubActionEncode(pSub);
if (pRedoRaw == NULL) return -1;
if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
return 0;
}
static int32_t mndCheckConsumer(SRpcMsg *pMsg, SHashObj* hash);
static int32_t mndSetSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);
@ -68,7 +64,7 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_VND_TMQ_SUBSCRIBE_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DELETE_SUB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DO_REBALANCE, mndProcessRebalanceReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessRebalanceReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP, mndProcessDropCgroupReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP_RSP, mndTransProcessRsp);
@ -213,16 +209,18 @@ static int32_t mndSplitSubscribeKey(const char *key, char *topic, char *cgroup,
}
static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
SMqRebInfo *pRebSub = taosHashGet(pHash, key, strlen(key) + 1);
if (pRebSub == NULL) {
pRebSub = tNewSMqRebSubscribe(key);
if (pRebSub == NULL) {
SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
if (pRebInfo == NULL) {
pRebInfo = tNewSMqRebSubscribe(key);
if (pRebInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
taosHashPut(pHash, key, strlen(key) + 1, pRebSub, sizeof(SMqRebInfo));
taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));
taosMemoryFree(pRebInfo);
pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
}
return pRebSub;
return pRebInfo;
}
static void doRemoveLostConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, const SMqRebInputObj *pInput) {
@ -727,17 +725,156 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
return 0;
}
static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
SMnode *pMnode = pMsg->info.node;
SMqDoRebalanceMsg *pReq = pMsg->pCont;
void *pIter = NULL;
// bool rebalanceOnce = false; // to ensure only once.
static void freeRebalanceItem(void *param) {
SMqRebInfo *pInfo = param;
taosArrayDestroy(pInfo->newConsumers);
taosArrayDestroy(pInfo->removedConsumers);
}
mInfo("mq re-balance start, total required re-balanced trans:%d", taosHashGetSize(pReq->rebSubHash));
static int32_t mndCheckConsumer(SRpcMsg *pMsg, SHashObj* rebSubHash) {
SMnode *pMnode = pMsg->info.node;
SSdb *pSdb = pMnode->pSdb;
SMqConsumerObj *pConsumer;
void *pIter = NULL;
mInfo("start to process mq timer");
// iterate all consumers, find all modification
while (1) {
pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
if (pIter == NULL) {
break;
}
int32_t hbStatus = atomic_add_fetch_32(&pConsumer->hbStatus, 1);
int32_t status = atomic_load_32(&pConsumer->status);
mInfo("check for consumer:0x%" PRIx64 " status:%d(%s), sub-time:%" PRId64 ", createTime:%" PRId64 ", hbstatus:%d",
pConsumer->consumerId, status, mndConsumerStatusName(status), pConsumer->subscribeTime, pConsumer->createTime,
hbStatus);
if (status == MQ_CONSUMER_STATUS_READY) {
if (taosArrayGetSize(pConsumer->assignedTopics) == 0) { // unsubscribe or close
mndDropConsumerFromSdb(pMnode, pConsumer->consumerId, &pMsg->info);
} else if (hbStatus > MND_CONSUMER_LOST_HB_CNT) {
taosRLockLatch(&pConsumer->lock);
int32_t topicNum = taosArrayGetSize(pConsumer->currentTopics);
for (int32_t i = 0; i < topicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *removedTopic = taosArrayGetP(pConsumer->currentTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(rebSubHash, key);
taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
}
taosRUnLockLatch(&pConsumer->lock);
}else{
int32_t newTopicNum = taosArrayGetSize(pConsumer->currentTopics);
for (int32_t i = 0; i < newTopicNum; i++) {
char * topic = taosArrayGetP(pConsumer->currentTopics, i);
SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic);
if (pSub == NULL) {
continue;
}
taosRLockLatch(&pSub->lock);
// 2.2 iterate all vg assigned to the consumer of that topic
SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));
int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);
for (int32_t j = 0; j < vgNum; j++) {
SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
SVgObj * pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId);
if (!pVgroup) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
mndMakeSubscribeKey(key, pConsumer->cgroup, topic);
mndGetOrCreateRebSub(rebSubHash, key);
mInfo("vnode splitted, vgId:%d rebalance will be triggered", pVgEp->vgId);
}
mndReleaseVgroup(pMnode, pVgroup);
}
taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub);
}
}
} else if (status == MQ_CONSUMER_STATUS_LOST) {
if (hbStatus > MND_CONSUMER_LOST_CLEAR_THRESHOLD) { // clear consumer if lost a day
mndDropConsumerFromSdb(pMnode, pConsumer->consumerId, &pMsg->info);
}
} else {
taosRLockLatch(&pConsumer->lock);
int32_t newTopicNum = taosArrayGetSize(pConsumer->rebNewTopics);
for (int32_t i = 0; i < newTopicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *newTopic = taosArrayGetP(pConsumer->rebNewTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, newTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(rebSubHash, key);
taosArrayPush(pRebSub->newConsumers, &pConsumer->consumerId);
}
int32_t removedTopicNum = taosArrayGetSize(pConsumer->rebRemovedTopics);
for (int32_t i = 0; i < removedTopicNum; i++) {
char key[TSDB_SUBSCRIBE_KEY_LEN];
char *removedTopic = taosArrayGetP(pConsumer->rebRemovedTopics, i);
mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
SMqRebInfo *pRebSub = mndGetOrCreateRebSub(rebSubHash, key);
taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
}
if (newTopicNum == 0 && removedTopicNum == 0 && taosArrayGetSize(pConsumer->assignedTopics) == 0) { // unsubscribe or close
mndDropConsumerFromSdb(pMnode, pConsumer->consumerId, &pMsg->info);
}
taosRUnLockLatch(&pConsumer->lock);
}
mndReleaseConsumer(pMnode, pConsumer);
}
return 0;
}
bool mndRebTryStart() {
int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1);
mInfo("rebalance counter old val:%d", old);
return old == 0;
}
void mndRebCntInc() {
int32_t val = atomic_add_fetch_32(&mqRebInExecCnt, 1);
mInfo("rebalance cnt inc, value:%d", val);
}
void mndRebCntDec() {
int32_t val = atomic_sub_fetch_32(&mqRebInExecCnt, 1);
mInfo("rebalance cnt sub, value:%d", val);
}
static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
int code = 0;
if (!mndRebTryStart()) {
mInfo("mq rebalance already in progress, do nothing");
return code;
}
SHashObj *rebSubHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK);
if (rebSubHash == NULL) {
mError("failed to create rebalance hashmap");
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = -1;
goto END;
}
taosHashSetFreeFp(rebSubHash, freeRebalanceItem);
mndCheckConsumer(pMsg, rebSubHash);
mInfo("mq re-balance start, total required re-balanced trans:%d", taosHashGetSize(rebSubHash));
// here we only handle one topic rebalance requirement to ensure the atomic execution of this transaction.
void *pIter = NULL;
SMnode *pMnode = pMsg->info.node;
while (1) {
pIter = taosHashIterate(pReq->rebSubHash, pIter);
pIter = taosHashIterate(rebSubHash, pIter);
if (pIter == NULL) {
break;
}
@ -756,12 +893,11 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
taosArrayDestroy(rebOutput.modifyConsumers);
taosArrayDestroy(rebOutput.rebVgs);
taosHashCancelIterate(pReq->rebSubHash, pIter);
taosHashCancelIterate(rebSubHash, pIter);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mInfo("mq re-balance failed, due to out of memory");
taosHashCleanup(pReq->rebSubHash);
mndRebEnd();
return -1;
mError("mq re-balance failed, due to out of memory");
code = -1;
goto END;
}
SMqRebInfo *pRebInfo = (SMqRebInfo *)pIter;
@ -829,10 +965,12 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
// reset flag
mInfo("mq re-balance completed successfully");
taosHashCleanup(pReq->rebSubHash);
mndRebEnd();
return 0;
END:
taosHashCleanup(rebSubHash);
mndRebCntDec();
return code;
}
static int32_t sendDeleteSubToVnode(SMqSubscribeObj *pSub, STrans *pTrans){

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

@ -15,7 +15,7 @@
#define _DEFAULT_SOURCE
#include "mndTrans.h"
#include "mndConsumer.h"
#include "mndSubscribe.h"
#include "mndDb.h"
#include "mndPrivilege.h"
#include "mndShow.h"

38
source/dnode/vnode/src/inc/tsdb.h
View File

@ -256,18 +256,7 @@ void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]);
// tsdbFS.c ==============================================================================================
int32_t tsdbFSOpen(STsdb *pTsdb, int8_t rollback);
int32_t tsdbFSClose(STsdb *pTsdb);
int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS);
void tsdbFSDestroy(STsdbFS *pFS);
int32_t tDFileSetCmprFn(const void *p1, const void *p2);
int32_t tsdbFSCommit(STsdb *pTsdb);
int32_t tsdbFSRollback(STsdb *pTsdb);
int32_t tsdbFSPrepareCommit(STsdb *pTsdb, STsdbFS *pFS);
int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS);
void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS);
void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t);
int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet);
int32_t tsdbFSUpsertDelFile(STsdbFS *pFS, SDelFile *pDelFile);
// tsdbReaderWriter.c ==============================================================================================
// SDataFWriter
int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet);
@ -737,7 +726,6 @@ struct STsdbReadSnap {
SMemTable *pIMem;
SQueryNode *pINode;
TFileSetArray *pfSetArray;
STsdbFS fs;
};
struct SDataFWriter {
@ -796,16 +784,16 @@ typedef struct {
} SSttTableRowsInfo;
typedef struct SSttBlockLoadInfo {
SBlockDataInfo blockData[2]; // buffered block data
SArray *aSttBlk;
int32_t currentLoadBlockIndex;
STSchema *pSchema;
int16_t *colIds;
int32_t numOfCols;
bool checkRemainingRow; // todo: no assign value?
bool isLast;
bool sttBlockLoaded;
SSttTableRowsInfo info;
SBlockDataInfo blockData[2]; // buffered block data
SArray *aSttBlk;
int32_t currentLoadBlockIndex;
STSchema *pSchema;
int16_t *colIds;
int32_t numOfCols;
bool checkRemainingRow; // todo: no assign value?
bool isLast;
bool sttBlockLoaded;
SSttTableRowsInfo info;
SSttBlockLoadCostInfo cost;
} SSttBlockLoadInfo;
@ -894,15 +882,15 @@ typedef struct {
_load_tomb_fn loadTombFn;
void *pReader;
void *idstr;
bool rspRows; // response the rows in stt-file, if possible
bool rspRows; // response the rows in stt-file, if possible
} SMergeTreeConf;
typedef struct SSttDataInfoForTable {
SArray* pTimeWindowList;
SArray *pTimeWindowList;
int64_t numOfRows;
} SSttDataInfoForTable;
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoForTable* pTableInfo);
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoForTable *pTableInfo);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
void tMergeTreePinSttBlock(SMergeTree *pMTree);

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

@ -95,7 +95,7 @@ typedef struct SQueryNode SQueryNode;
#define VNODE_RSMA2_DIR "rsma2"
#define VNODE_TQ_STREAM "stream"
#if SUSPEND_RESUME_TEST // only for test purpose
#if SUSPEND_RESUME_TEST // only for test purpose
#define VNODE_BUFPOOL_SEGMENTS 1
#else
#define VNODE_BUFPOOL_SEGMENTS 3
@ -216,8 +216,6 @@ int32_t tsdbBegin(STsdb* pTsdb);
int32_t tsdbCacheCommit(STsdb* pTsdb);
int32_t tsdbCompact(STsdb* pTsdb, SCompactInfo* pInfo);
int32_t tsdbRetention(STsdb* tsdb, int64_t now, int32_t sync);
// int32_t tsdbFinishCommit(STsdb* pTsdb);
// int32_t tsdbRollbackCommit(STsdb* pTsdb);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp);
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows);

5
source/dnode/vnode/src/tq/tq.c
View File

@ -1292,10 +1292,9 @@ int32_t tqProcessTaskCheckPointSourceReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg* pRsp)
// check if the checkpoint msg already sent or not.
if (status == TASK_STATUS__CK) {
ASSERT(pTask->chkInfo.checkpointingId == req.checkpointId);
tqWarn("s-task:%s recv checkpoint-source msg again checkpointId:%" PRId64
" already received, ignore this msg and continue process checkpoint",
pTask->id.idStr, pTask->chkInfo.checkpointingId);
" transId:%d already received, ignore this msg and continue process checkpoint",
pTask->id.idStr, pTask->chkInfo.checkpointingId, req.transId);
taosThreadMutexUnlock(&pTask->lock);
streamMetaReleaseTask(pMeta, pTask);

8
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -1183,8 +1183,10 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen);
if (h) {
SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pTsdb->lruCache, h);
if (pLastCol->dirty && (pLastCol->ts <= eKey && pLastCol->ts >= sKey)) {
if (pLastCol->dirty) {
pLastCol->dirty = 0;
}
if (pLastCol->ts <= eKey && pLastCol->ts >= sKey) {
erase = true;
}
taosLRUCacheRelease(pTsdb->lruCache, h, erase);
@ -1197,8 +1199,10 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[num_keys + i], klen);
if (h) {
SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pTsdb->lruCache, h);
if (pLastCol->dirty && (pLastCol->ts <= eKey && pLastCol->ts >= sKey)) {
if (pLastCol->dirty) {
pLastCol->dirty = 0;
}
if (pLastCol->ts <= eKey && pLastCol->ts >= sKey) {
erase = true;
}
taosLRUCacheRelease(pTsdb->lruCache, h, erase);

1628
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

File diff suppressed because it is too large Load Diff

313
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -702,7 +702,7 @@ _exit:
}
// EXPOSED APIS ====================================================================================
int32_t tsdbFSCommit(STsdb *pTsdb) {
static int32_t tsdbFSCommit(STsdb *pTsdb) {
int32_t code = 0;
int32_t lino = 0;
STsdbFS fs = {0};
@ -738,7 +738,7 @@ _exit:
return code;
}
int32_t tsdbFSRollback(STsdb *pTsdb) {
static int32_t tsdbFSRollback(STsdb *pTsdb) {
int32_t code = 0;
int32_t lino = 0;
@ -833,312 +833,3 @@ int32_t tsdbFSClose(STsdb *pTsdb) {
return code;
}
int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS) {
int32_t code = 0;
int32_t lino = 0;
pFS->pDelFile = NULL;
if (pFS->aDFileSet) {
taosArrayClear(pFS->aDFileSet);
} else {
pFS->aDFileSet = taosArrayInit(taosArrayGetSize(pTsdb->fs.aDFileSet), sizeof(SDFileSet));
if (pFS->aDFileSet == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
}
if (pTsdb->fs.pDelFile) {
pFS->pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
if (pFS->pDelFile == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
*pFS->pDelFile = *pTsdb->fs.pDelFile;
}
for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);
SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid};
// head
fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
if (fSet.pHeadF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
*fSet.pHeadF = *pSet->pHeadF;
// data
fSet.pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
if (fSet.pDataF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
*fSet.pDataF = *pSet->pDataF;
// sma
fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
if (fSet.pSmaF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
*fSet.pSmaF = *pSet->pSmaF;
// stt
for (fSet.nSttF = 0; fSet.nSttF < pSet->nSttF; fSet.nSttF++) {
fSet.aSttF[fSet.nSttF] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
if (fSet.aSttF[fSet.nSttF] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
*fSet.aSttF[fSet.nSttF] = *pSet->aSttF[fSet.nSttF];
}
if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
}
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
}
return code;
}
int32_t tsdbFSUpsertDelFile(STsdbFS *pFS, SDelFile *pDelFile) {
int32_t code = 0;
if (pDelFile) {
if (pFS->pDelFile == NULL) {
pFS->pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));
if (pFS->pDelFile == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
*pFS->pDelFile = *pDelFile;
} else {
if (pFS->pDelFile) {
taosMemoryFree(pFS->pDelFile);
pFS->pDelFile = NULL;
}
}
_exit:
return code;
}
int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
int32_t code = 0;
int32_t idx = taosArraySearchIdx(pFS->aDFileSet, pSet, tDFileSetCmprFn, TD_GE);
if (idx < 0) {
idx = taosArrayGetSize(pFS->aDFileSet);
} else {
SDFileSet *pDFileSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, idx);
int32_t c = tDFileSetCmprFn(pSet, pDFileSet);
if (c == 0) {
*pDFileSet->pHeadF = *pSet->pHeadF;
*pDFileSet->pDataF = *pSet->pDataF;
*pDFileSet->pSmaF = *pSet->pSmaF;
// stt
if (pSet->nSttF > pDFileSet->nSttF) {
ASSERT(pSet->nSttF == pDFileSet->nSttF + 1);
pDFileSet->aSttF[pDFileSet->nSttF] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
if (pDFileSet->aSttF[pDFileSet->nSttF] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*pDFileSet->aSttF[pDFileSet->nSttF] = *pSet->aSttF[pSet->nSttF - 1];
pDFileSet->nSttF++;
} else if (pSet->nSttF < pDFileSet->nSttF) {
ASSERT(pSet->nSttF == 1);
for (int32_t iStt = 1; iStt < pDFileSet->nSttF; iStt++) {
taosMemoryFree(pDFileSet->aSttF[iStt]);
}
*pDFileSet->aSttF[0] = *pSet->aSttF[0];
pDFileSet->nSttF = 1;
} else {
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
*pDFileSet->aSttF[iStt] = *pSet->aSttF[iStt];
}
}
pDFileSet->diskId = pSet->diskId;
goto _exit;
}
}
ASSERT(pSet->nSttF == 1);
SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid, .nSttF = 1};
// head
fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
if (fSet.pHeadF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.pHeadF = *pSet->pHeadF;
// data
fSet.pDataF = (SDataFile *)taosMemoryMalloc(sizeof(SDataFile));
if (fSet.pDataF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.pDataF = *pSet->pDataF;
// sma
fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
if (fSet.pSmaF == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.pSmaF = *pSet->pSmaF;
// stt
fSet.aSttF[0] = (SSttFile *)taosMemoryMalloc(sizeof(SSttFile));
if (fSet.aSttF[0] == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
*fSet.aSttF[0] = *pSet->aSttF[0];
if (taosArrayInsert(pFS->aDFileSet, idx, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
_exit:
return code;
}
int32_t tsdbFSPrepareCommit(STsdb *pTsdb, STsdbFS *pFSNew) {
int32_t code = 0;
int32_t lino = 0;
char tfname[TSDB_FILENAME_LEN];
tsdbGetCurrentFName(pTsdb, NULL, tfname);
// gnrt CURRENT.t
code = tsdbSaveFSToFile(pFSNew, tfname);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
}
return code;
}
int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS) {
int32_t code = 0;
int32_t nRef;
pFS->aDFileSet = taosArrayInit(taosArrayGetSize(pTsdb->fs.aDFileSet), sizeof(SDFileSet));
if (pFS->aDFileSet == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pFS->pDelFile = pTsdb->fs.pDelFile;
if (pFS->pDelFile) {
nRef = atomic_fetch_add_32(&pFS->pDelFile->nRef, 1);
ASSERT(nRef > 0);
}
SDFileSet fSet;
for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
SDFileSet *pSet = (SDFileSet *)taosArrayGet(pTsdb->fs.aDFileSet, iSet);
fSet = *pSet;
nRef = atomic_fetch_add_32(&pSet->pHeadF->nRef, 1);
ASSERT(nRef > 0);
nRef = atomic_fetch_add_32(&pSet->pDataF->nRef, 1);
ASSERT(nRef > 0);
nRef = atomic_fetch_add_32(&pSet->pSmaF->nRef, 1);
ASSERT(nRef > 0);
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
nRef = atomic_fetch_add_32(&pSet->aSttF[iStt]->nRef, 1);
ASSERT(nRef > 0);
}
if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
_exit:
return code;
}
void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS) {
int32_t nRef;
char fname[TSDB_FILENAME_LEN];
if (pFS->pDelFile) {
nRef = atomic_sub_fetch_32(&pFS->pDelFile->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbDelFileName(pTsdb, pFS->pDelFile, fname);
(void)taosRemoveFile(fname);
taosMemoryFree(pFS->pDelFile);
}
}
for (int32_t iSet = 0; iSet < taosArrayGetSize(pFS->aDFileSet); iSet++) {
SDFileSet *pSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet);
// head
nRef = atomic_sub_fetch_32(&pSet->pHeadF->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
(void)taosRemoveFile(fname);
taosMemoryFree(pSet->pHeadF);
}
// data
nRef = atomic_sub_fetch_32(&pSet->pDataF->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
(void)taosRemoveFile(fname);
taosMemoryFree(pSet->pDataF);
}
// sma
nRef = atomic_sub_fetch_32(&pSet->pSmaF->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
(void)taosRemoveFile(fname);
taosMemoryFree(pSet->pSmaF);
}
// stt
for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
nRef = atomic_sub_fetch_32(&pSet->aSttF[iStt]->nRef, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
tsdbSttFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSttF[iStt], fname);
(void)taosRemoveFile(fname);
taosMemoryFree(pSet->aSttF[iStt]);
/* code */
}
}
}
taosArrayDestroy(pFS->aDFileSet);
}

6
source/dnode/vnode/src/tsdb/tsdbFS2.c
View File

@ -805,6 +805,12 @@ int64_t tsdbFSAllocEid(STFileSystem *fs) {
return cid;
}
void tsdbFSUpdateEid(STFileSystem *fs, int64_t cid) {
taosThreadMutexLock(&fs->tsdb->mutex);
fs->neid = TMAX(fs->neid, cid);
taosThreadMutexUnlock(&fs->tsdb->mutex);
}
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype) {
int32_t code = 0;
int32_t lino;

1
source/dnode/vnode/src/tsdb/tsdbFS2.h
View File

@ -52,6 +52,7 @@ int32_t tsdbFSCreateRefRangedSnapshot(STFileSystem *fs, int64_t sver, int64_t ev
int32_t tsdbFSDestroyRefRangedSnapshot(TFileSetRangeArray **fsrArr);
// txn
int64_t tsdbFSAllocEid(STFileSystem *fs);
void tsdbFSUpdateEid(STFileSystem *fs, int64_t cid);
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype);
int32_t tsdbFSEditCommit(STFileSystem *fs);
int32_t tsdbFSEditAbort(STFileSystem *fs);

6
source/dnode/vnode/src/tsdb/tsdbFSetRAW.c
View File

@ -14,6 +14,7 @@
*/
#include "tsdbFSetRAW.h"
#include "tsdbFS2.h"
// SFSetRAWWriter ==================================================
typedef struct SFSetRAWWriter {
@ -76,7 +77,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
.szPage = writer->config->szPage,
.fid = bHdr->file.fid,
.did = writer->config->did,
.cid = writer->config->cid,
.cid = bHdr->file.cid,
.level = writer->config->level,
.file =
@ -84,7 +85,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
.type = bHdr->file.type,
.fid = bHdr->file.fid,
.did = writer->config->did,
.cid = writer->config->cid,
.cid = bHdr->file.cid,
.size = bHdr->file.size,
.minVer = bHdr->file.minVer,
.maxVer = bHdr->file.maxVer,
@ -94,6 +95,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
},
};
tsdbFSUpdateEid(config.tsdb->pFS, config.cid);
writer->ctx->offset = 0;
writer->ctx->file = config.file;

119
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -22,8 +22,8 @@
#include "tsdbUtil2.h"
#include "tsimplehash.h"
#define ASCENDING_TRAVERSE(o) (o == TSDB_ORDER_ASC)
#define getCurrentKeyInSttBlock(_r) ((_r)->currentKey)
#define ASCENDING_TRAVERSE(o) (o == TSDB_ORDER_ASC)
#define getCurrentKeyInSttBlock(_r) ((_r)->currentKey)
typedef struct {
bool overlapWithNeighborBlock;
@ -41,7 +41,7 @@ static int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, i
static TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* pReader);
static int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
static int32_t doMergeRowsInSttBlock(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
SRowMerger* pMerger, SVersionRange* pVerRange, const char* id);
SRowMerger* pMerger, SVersionRange* pVerRange, const char* id);
static int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, STsdbReader* pReader);
static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, SRow* pTSRow,
STableBlockScanInfo* pScanInfo);
@ -67,7 +67,7 @@ static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond
static int32_t doBuildDataBlock(STsdbReader* pReader);
static TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
static bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
static bool hasDataInSttBlock(STableBlockScanInfo *pInfo);
static bool hasDataInSttBlock(STableBlockScanInfo* pInfo);
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter);
static int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order);
static void resetTableListIndex(SReaderStatus* pStatus);
@ -1138,7 +1138,7 @@ static bool getNeighborBlockOfSameTable(SDataBlockIter* pBlockIter, SFileDataBlo
return false;
}
int32_t step = asc ? 1 : -1;
int32_t step = asc ? 1 : -1;
STableDataBlockIdx* pTableDataBlockIdx =
taosArrayGet(pTableBlockScanInfo->pBlockIdxList, pBlockInfo->tbBlockIdx + step);
SFileDataBlockInfo* p = taosArrayGet(pBlockIter->blockList, pTableDataBlockIdx->globalIndex);
@ -1316,17 +1316,17 @@ static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo*
return TSDB_CODE_SUCCESS;
}
int64_t st = taosGetTimestampUs();
int64_t st = taosGetTimestampUs();
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->resBlockInfo.capacity, pReader);
int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->resBlockInfo.capacity, pReader);
double el = (taosGetTimestampUs() - st) / 1000.0;
updateComposedBlockInfo(pReader, el, pBlockScanInfo);
tsdbDebug("%p build data block from cache completed, elapsed time:%.2f ms, numOfRows:%" PRId64 ", brange:%" PRId64
" - %" PRId64 ", uid:%" PRIu64 ", %s",
pReader, el, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
pBlockScanInfo->uid, pReader->idStr);
pReader, el, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey, pBlockScanInfo->uid,
pReader->idStr);
pReader->cost.buildmemBlock += el;
return code;
@ -1390,13 +1390,9 @@ static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockSc
}
}
static void doPinSttBlock(SSttBlockReader* pSttBlockReader) {
tMergeTreePinSttBlock(&pSttBlockReader->mergeTree);
}
static void doPinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreePinSttBlock(&pSttBlockReader->mergeTree); }
static void doUnpinSttBlock(SSttBlockReader* pSttBlockReader) {
tMergeTreeUnpinSttBlock(&pSttBlockReader->mergeTree);
}
static void doUnpinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreeUnpinSttBlock(&pSttBlockReader->mergeTree); }
static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SSttBlockReader* pSttBlockReader,
STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader,
@ -1535,8 +1531,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange,
pReader->idStr);
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr);
}
if (minKey == k.ts) {
@ -1585,8 +1580,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange,
pReader->idStr);
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr);
}
if (minKey == key) {
@ -1648,7 +1642,7 @@ static int32_t doMergeFileBlockAndLastBlock(SSttBlockReader* pSttBlockReader, ST
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
tsdbRowMergerAdd(pMerger, pRow1, NULL);
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->info.verRange,
pReader->idStr);
pReader->idStr);
code = tsdbRowMergerGetRow(pMerger, &pTSRow);
if (code != TSDB_CODE_SUCCESS) {
@ -1671,7 +1665,7 @@ static int32_t doMergeFileBlockAndLastBlock(SSttBlockReader* pSttBlockReader, ST
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->info.verRange,
pReader->idStr);
pReader->idStr);
// merge with block data if ts == key
if (tsLastBlock == pBlockData->aTSKEY[pDumpInfo->rowIndex]) {
@ -1740,7 +1734,7 @@ static int32_t mergeFileBlockAndSttBlock(STsdbReader* pReader, SSttBlockReader*
// the following for key == tsLast
// ASC: file block ------> stt block
// DESC: stt block ------> file block
SRow* pTSRow = NULL;
SRow* pTSRow = NULL;
if (ASCENDING_TRAVERSE(pReader->info.order)) {
code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
if (code != TSDB_CODE_SUCCESS) {
@ -1889,8 +1883,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange,
pReader->idStr);
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr);
}
if (minKey == ik.ts) {
@ -1948,8 +1941,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange,
pReader->idStr);
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr);
}
if (minKey == key) {
@ -2120,7 +2112,7 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
};
SSttDataInfoForTable info = {.pTimeWindowList = taosArrayInit(4, sizeof(STimeWindow))};
int32_t code = tMergeTreeOpen2(&pSttBlockReader->mergeTree, &conf, &info);
int32_t code = tMergeTreeOpen2(&pSttBlockReader->mergeTree, &conf, &info);
if (code != TSDB_CODE_SUCCESS) {
return false;
}
@ -2138,7 +2130,7 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
pScanInfo->sttWindow.ekey = INT64_MIN;
// calculate the time window for data in stt files
for(int32_t i = 0; i < taosArrayGetSize(info.pTimeWindowList); ++i) {
for (int32_t i = 0; i < taosArrayGetSize(info.pTimeWindowList); ++i) {
STimeWindow* pWindow = taosArrayGet(info.pTimeWindowList, i);
if (pScanInfo->sttWindow.skey > pWindow->skey) {
pScanInfo->sttWindow.skey = pWindow->skey;
@ -2149,8 +2141,9 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
}
}
pScanInfo->sttKeyInfo.status = taosArrayGetSize(info.pTimeWindowList)? STT_FILE_HAS_DATA:STT_FILE_NO_DATA;
pScanInfo->sttKeyInfo.nextProcKey = ASCENDING_TRAVERSE(pReader->info.order)? pScanInfo->sttWindow.skey:pScanInfo->sttWindow.ekey;
pScanInfo->sttKeyInfo.status = taosArrayGetSize(info.pTimeWindowList) ? STT_FILE_HAS_DATA : STT_FILE_NO_DATA;
pScanInfo->sttKeyInfo.nextProcKey =
ASCENDING_TRAVERSE(pReader->info.order) ? pScanInfo->sttWindow.skey : pScanInfo->sttWindow.ekey;
hasData = true;
} else {
hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, &pReader->info.verRange);
@ -2168,9 +2161,7 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
return hasData;
}
static bool hasDataInSttBlock(STableBlockScanInfo *pInfo) {
return pInfo->sttKeyInfo.status == STT_FILE_HAS_DATA;
}
static bool hasDataInSttBlock(STableBlockScanInfo* pInfo) { return pInfo->sttKeyInfo.status == STT_FILE_HAS_DATA; }
bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) {
if ((pBlockData->nRow > 0) && (pBlockData->nRow != pDumpInfo->totalRows)) {
@ -2225,10 +2216,11 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc
}
}
int32_t mergeRowsInSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
bool copied = false;
SRow* pTSRow = NULL;
int64_t tsLastBlock = getCurrentKeyInSttBlock(pSttBlockReader);
int32_t mergeRowsInSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pBlockScanInfo,
STsdbReader* pReader) {
bool copied = false;
SRow* pTSRow = NULL;
int64_t tsLastBlock = getCurrentKeyInSttBlock(pSttBlockReader);
SRowMerger* pMerger = &pReader->status.merger;
TSDBROW* pRow = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
@ -2559,16 +2551,16 @@ static void prepareDurationForNextFileSet(STsdbReader* pReader) {
pReader->status.bProcMemFirstFileset = false;
}
int32_t fid = pReader->status.pCurrentFileset->fid;
int32_t fid = pReader->status.pCurrentFileset->fid;
STimeWindow winFid = {0};
tsdbFidKeyRange(fid, pReader->pTsdb->keepCfg.days, pReader->pTsdb->keepCfg.precision, &winFid.skey, &winFid.ekey);
if (ASCENDING_TRAVERSE(pReader->info.order)) {
pReader->status.bProcMemPreFileset = !(pReader->status.memTableMaxKey < pReader->status.prevFilesetStartKey ||
(winFid.skey-1) < pReader->status.memTableMinKey);
(winFid.skey - 1) < pReader->status.memTableMinKey);
} else {
pReader->status.bProcMemPreFileset = !( pReader->status.memTableMaxKey < (winFid.ekey+1) ||
pReader->status.prevFilesetEndKey < pReader->status.memTableMinKey);
pReader->status.bProcMemPreFileset = !(pReader->status.memTableMaxKey < (winFid.ekey + 1) ||
pReader->status.prevFilesetEndKey < pReader->status.memTableMinKey);
}
if (pReader->status.bProcMemPreFileset) {
@ -2802,7 +2794,7 @@ static int32_t doLoadSttBlockSequentially(STsdbReader* pReader) {
static bool notOverlapWithSttFiles(SFileDataBlockInfo* pBlockInfo, STableBlockScanInfo* pScanInfo, bool asc) {
ASSERT(pScanInfo->sttKeyInfo.status != STT_FILE_READER_UNINIT);
if(pScanInfo->sttKeyInfo.status == STT_FILE_NO_DATA) {
if (pScanInfo->sttKeyInfo.status == STT_FILE_NO_DATA) {
return true;
} else {
int64_t keyInStt = pScanInfo->sttKeyInfo.nextProcKey;
@ -2923,7 +2915,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
static int32_t buildBlockFromBufferSeqForPreFileset(STsdbReader* pReader, int64_t endKey) {
SReaderStatus* pStatus = &pReader->status;
tsdbDebug("seq load data blocks from cache that preceeds fileset %d, %s", pReader->status.pCurrentFileset->fid, pReader->idStr);
tsdbDebug("seq load data blocks from cache that preceeds fileset %d, %s", pReader->status.pCurrentFileset->fid,
pReader->idStr);
while (1) {
if (pReader->code != TSDB_CODE_SUCCESS) {
@ -3847,7 +3840,6 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e
pBlockScanInfo->lastProcKey = row.pBlockData->aTSKEY[row.iRow];
}
// no data in buffer, return immediately
if (!(pBlockScanInfo->iter.hasVal || pBlockScanInfo->iiter.hasVal)) {
break;
@ -3945,7 +3937,7 @@ static int32_t doOpenReaderImpl(STsdbReader* pReader) {
int32_t code = TSDB_CODE_SUCCESS;
if (pStatus->fileIter.numOfFiles == 0) {
pStatus->loadFromFile = false;
// } else if (READER_EXEC_DATA == pReader->info.readMode) {
// } else if (READER_EXEC_DATA == pReader->info.readMode) {
// DO NOTHING
} else {
code = initForFirstBlockInFile(pReader, pBlockIter);
@ -4092,7 +4084,7 @@ int32_t tsdbReaderOpen2(void* pVnode, SQueryTableDataCond* pCond, void* pTableLi
}
pReader->flag = READER_STATUS_SUSPEND;
pReader->info.execMode = pCond->notLoadData? READER_EXEC_ROWS : READER_EXEC_DATA;
pReader->info.execMode = pCond->notLoadData ? READER_EXEC_ROWS : READER_EXEC_DATA;
pReader->pIgnoreTables = pIgnoreTables;
tsdbDebug("%p total numOfTable:%d, window:%" PRId64 " - %" PRId64 ", verRange:%" PRId64 " - %" PRId64
@ -4157,7 +4149,7 @@ void tsdbReaderClose2(STsdbReader* pReader) {
}
SReadCostSummary* pCost = &pReader->cost;
SFilesetIter* pFilesetIter = &pReader->status.fileIter;
SFilesetIter* pFilesetIter = &pReader->status.fileIter;
if (pFilesetIter->pSttBlockReader != NULL) {
SSttBlockReader* pLReader = pFilesetIter->pSttBlockReader;
tMergeTreeClose(&pLReader->mergeTree);
@ -4204,7 +4196,7 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status;
STableBlockScanInfo* pBlockScanInfo = NULL;
pReader->status.suspendInvoked = true; // record the suspend status
pReader->status.suspendInvoked = true; // record the suspend status
if (pStatus->loadFromFile) {
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
@ -4227,7 +4219,7 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
// resetDataBlockScanInfo excluding lastKey
STableBlockScanInfo** p = NULL;
int32_t step = ASCENDING_TRAVERSE(pReader->info.order)? 1:-1;
int32_t step = ASCENDING_TRAVERSE(pReader->info.order) ? 1 : -1;
int32_t iter = 0;
while ((p = tSimpleHashIterate(pStatus->pTableMap, p, &iter)) != NULL) {
@ -4248,7 +4240,7 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
pReader->flag = READER_STATUS_SUSPEND;
#if SUSPEND_RESUME_TEST
tsem_post(&pReader->resumeAfterSuspend);
tsem_post(&pReader->resumeAfterSuspend);
#endif
tsdbDebug("reader: %p suspended uid %" PRIu64 " in this query %s", pReader, pBlockScanInfo ? pBlockScanInfo->uid : 0,
@ -4331,13 +4323,13 @@ _err:
}
static int32_t buildFromPreFilesetBuffer(STsdbReader* pReader) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t code = TSDB_CODE_SUCCESS;
SReaderStatus* pStatus = &pReader->status;
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int32_t fid = pReader->status.pCurrentFileset->fid;
STimeWindow win = {0};
int32_t fid = pReader->status.pCurrentFileset->fid;
STimeWindow win = {0};
tsdbFidKeyRange(fid, pReader->pTsdb->keepCfg.days, pReader->pTsdb->keepCfg.precision, &win.skey, &win.ekey);
int64_t endKey = (ASCENDING_TRAVERSE(pReader->info.order)) ? win.skey : win.ekey;
@ -4359,8 +4351,8 @@ static int32_t buildFromPreFilesetBuffer(STsdbReader* pReader) {
static int32_t doTsdbNextDataBlockFilesetDelimited(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status;
int32_t code = TSDB_CODE_SUCCESS;
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int32_t code = TSDB_CODE_SUCCESS;
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
if (pStatus->loadFromFile) {
if (pStatus->bProcMemPreFileset) {
@ -4375,12 +4367,12 @@ static int32_t doTsdbNextDataBlockFilesetDelimited(STsdbReader* pReader) {
return code;
}
tsdbTrace("block from file rows: %"PRId64", will process pre-file set buffer: %d. %s",
pBlock->info.rows, pStatus->bProcMemFirstFileset, pReader->idStr);
tsdbTrace("block from file rows: %" PRId64 ", will process pre-file set buffer: %d. %s", pBlock->info.rows,
pStatus->bProcMemFirstFileset, pReader->idStr);
if (pStatus->bProcMemPreFileset) {
if (pBlock->info.rows > 0) {
if (pReader->notifyFn) {
int32_t fid = pReader->status.pCurrentFileset->fid;
int32_t fid = pReader->status.pCurrentFileset->fid;
STsdReaderNotifyInfo info = {0};
info.duration.filesetId = fid;
pReader->notifyFn(TSD_READER_NOTIFY_NEXT_DURATION_BLOCK, &info, pReader->notifyParam);
@ -4404,8 +4396,8 @@ static int32_t doTsdbNextDataBlockFilesetDelimited(STsdbReader* pReader) {
static int32_t doTsdbNextDataBlockFilesFirst(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status;
int32_t code = TSDB_CODE_SUCCESS;
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int32_t code = TSDB_CODE_SUCCESS;
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
if (pStatus->loadFromFile) {
code = buildBlockFromFiles(pReader);
@ -4940,7 +4932,7 @@ static void getMemTableTimeRange(STsdbReader* pReader, int64_t* pMaxKey, int64_t
int64_t minKey = INT64_MAX;
void* pHashIter = tSimpleHashIterate(pStatus->pTableMap, NULL, &iter);
while (pHashIter!= NULL) {
while (pHashIter != NULL) {
STableBlockScanInfo* pBlockScanInfo = *(STableBlockScanInfo**)pHashIter;
STbData* d = NULL;
@ -5144,7 +5136,6 @@ void tsdbUntakeReadSnap2(STsdbReader* pReader, STsdbReadSnap* pSnap, bool proact
tsdbUnrefMemTable(pSnap->pIMem, pSnap->pINode, proactive);
}
tsdbFSUnref(pTsdb, &pSnap->fs);
if (pSnap->pNode) taosMemoryFree(pSnap->pNode);
if (pSnap->pINode) taosMemoryFree(pSnap->pINode);
@ -5165,9 +5156,7 @@ void tsdbReaderSetId2(STsdbReader* pReader, const char* idstr) {
void tsdbReaderSetCloseFlag(STsdbReader* pReader) { /*pReader->code = TSDB_CODE_TSC_QUERY_CANCELLED;*/
}
void tsdbSetFilesetDelimited(STsdbReader* pReader) {
pReader->bFilesetDelimited = true;
}
void tsdbSetFilesetDelimited(STsdbReader* pReader) { pReader->bFilesetDelimited = true; }
void tsdbReaderSetNotifyCb(STsdbReader* pReader, TsdReaderNotifyCbFn notifyFn, void* param) {
pReader->notifyFn = notifyFn;

1
source/libs/executor/inc/executorInt.h
View File

@ -852,6 +852,7 @@ void resetWinRange(STimeWindow* winRange);
bool checkExpiredData(SStateStore* pAPI, SUpdateInfo* pUpdateInfo, STimeWindowAggSupp* pTwSup, uint64_t tableId, TSKEY ts);
int64_t getDeleteMark(SWindowPhysiNode* pWinPhyNode, int64_t interval);
void resetUnCloseSessionWinInfo(SSHashObj* winMap);
void setStreamOperatorCompleted(struct SOperatorInfo* pOperator);
int32_t encodeSSessionKey(void** buf, SSessionKey* key);
void* decodeSSessionKey(void* buf, SSessionKey* key);

7
source/libs/executor/src/executil.c
View File

@ -2244,8 +2244,11 @@ void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr
}
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
if (!pBlock || pBlock->info.rows == 0) {
qDebug("%s===stream===%s: Block is Null or Empty", taskIdStr, flag);
if (!pBlock) {
qDebug("%s===stream===%s: Block is Null", taskIdStr, flag);
return;
} else if (pBlock->info.rows == 0) {
qDebug("%s===stream===%s: Block is Empty. type:%d", taskIdStr, flag, pBlock->info.type);
return;
}
if (qDebugFlag & DEBUG_DEBUG) {

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

@ -1997,7 +1997,6 @@ void streamScanOperatorSaveCheckpoint(SStreamScanInfo* pInfo) {
int32_t len = streamScanOperatorEncode(pInfo, &pBuf);
pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), pBuf, len);
taosMemoryFree(pBuf);
pInfo->stateStore.streamStateCommit(pInfo->pState);
}
// other properties are recovered from the execution plan
@ -2316,6 +2315,7 @@ FETCH_NEXT_BLOCK:
doCheckUpdate(pInfo, pBlockInfo->window.ekey, pInfo->pRes);
doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
int64_t numOfUpdateRes = pInfo->pUpdateDataRes->info.rows;
qDebug("%s %" PRId64 " rows in datablock, update res:%" PRId64, id, pBlockInfo->rows, numOfUpdateRes);

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

@ -480,18 +480,18 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
return resBlock;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -561,7 +561,7 @@ static SSDataBlock* doStreamEventAgg(SOperatorInfo* pOperator) {
if (resBlock != NULL) {
return resBlock;
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}

80
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -801,10 +801,24 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
SRowBuffPos* pResPos = NULL;
SResultRow* pResult = NULL;
int32_t forwardRows = 0;
int32_t endRowId = pSDataBlock->info.rows - 1;
SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
tsCols = (int64_t*)pColDataInfo->pData;
if (pSDataBlock->info.window.skey != tsCols[0] || pSDataBlock->info.window.ekey != tsCols[endRowId]) {
qError("table uid %" PRIu64 " data block timestamp range may not be calculated! minKey %" PRId64
",maxKey %" PRId64,
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
blockDataUpdateTsWindow(pSDataBlock, pInfo->primaryTsIndex);
// timestamp of the data is incorrect
if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
qError("table uid %" PRIu64 " data block timestamp is out of range! minKey %" PRId64 ",maxKey %" PRId64,
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
}
}
int32_t startPos = 0;
TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
STimeWindow nextWin = {0};
@ -902,19 +916,6 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
pInfo->delKey = key;
}
int32_t prevEndPos = (forwardRows - 1) * step + startPos;
if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
qError("table uid %" PRIu64 " data block timestamp range may not be calculated! minKey %" PRId64
",maxKey %" PRId64,
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
blockDataUpdateTsWindow(pSDataBlock, 0);
// timestamp of the data is incorrect
if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
qError("table uid %" PRIu64 " data block timestamp is out of range! minKey %" PRId64 ",maxKey %" PRId64,
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
}
}
if (IS_FINAL_INTERVAL_OP(pOperator)) {
startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos);
} else {
@ -1223,7 +1224,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
return pInfo->pCheckpointRes;
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
if (!IS_FINAL_INTERVAL_OP(pOperator)) {
clearFunctionContext(&pOperator->exprSupp);
// semi interval operator clear disk buffer
@ -2343,10 +2344,6 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, void* pState, SSDa
}
int32_t code = pAPI->stateStore.streamStateGetByPos(pState, pPos, (void**)&pRow);
if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
ASSERT(pBlock->info.rows > 0);
break;
}
if (code == -1) {
// for history
@ -2363,6 +2360,11 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, void* pState, SSDa
continue;
}
if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
ASSERT(pBlock->info.rows > 0);
break;
}
pGroupResInfo->index += 1;
for (int32_t j = 0; j < numOfExprs; ++j) {
@ -2609,18 +2611,18 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
return opRes;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -2718,7 +2720,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
return opRes;
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -3001,7 +3003,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
clearFunctionContext(&pOperator->exprSupp);
// semi session operator clear disk buffer
clearStreamSessionOperator(pInfo);
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
pInfo->clearState = false;
return NULL;
}
@ -3074,7 +3076,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
clearFunctionContext(&pOperator->exprSupp);
// semi session operator clear disk buffer
clearStreamSessionOperator(pInfo);
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -3551,18 +3553,18 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
return resBlock;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
}
if (pInfo->recvGetAll) {
pInfo->recvGetAll = false;
resetUnCloseSessionWinInfo(pInfo->streamAggSup.pResultRows);
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -3628,7 +3630,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
if (resBlock != NULL) {
return resBlock;
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -3864,11 +3866,11 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
if (pInfo->reCkBlock) {
pInfo->reCkBlock = false;
// printDataBlock(pInfo->pCheckpointRes, "single interval ck");
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
return pInfo->pCheckpointRes;
}
setOperatorCompleted(pOperator);
setStreamOperatorCompleted(pOperator);
return NULL;
}
@ -3900,7 +3902,6 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
doDeleteWindows(pOperator, &pInfo->interval, pBlock, pInfo->pDelWins, pInfo->pUpdatedMap);
continue;
} else if (pBlock->info.type == STREAM_GET_ALL) {
qDebug("===stream===%s recv|block type STREAM_GET_ALL", getStreamOpName(pOperator->operatorType));
pInfo->recvGetAll = true;
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
continue;
@ -4083,3 +4084,8 @@ _error:
pTaskInfo->code = code;
return NULL;
}
void setStreamOperatorCompleted(SOperatorInfo* pOperator) {
setOperatorCompleted(pOperator);
qDebug("stask:%s %s status: %d. set completed", GET_TASKID(pOperator->pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status);
}

4
source/libs/parser/src/parTranslater.c
View File

@ -8941,7 +8941,7 @@ static int32_t extractShowVariablesResultSchema(int32_t* numOfCols, SSchema** pS
(*pSchema)[0].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[0].bytes = TSDB_CONFIG_OPTION_LEN;
strcpy((*pSchema)[0].name, "result");
strcpy((*pSchema)[0].name, "name");
(*pSchema)[1].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[1].bytes = TSDB_CONFIG_VALUE_LEN;
@ -8963,7 +8963,7 @@ static int32_t extractCompactDbResultSchema(int32_t* numOfCols, SSchema** pSchem
(*pSchema)[0].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[0].bytes = COMPACT_DB_RESULT_FIELD1_LEN;
strcpy((*pSchema)[0].name, "name");
strcpy((*pSchema)[0].name, "result");
(*pSchema)[1].type = TSDB_DATA_TYPE_INT;
(*pSchema)[1].bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;

28
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -665,6 +665,7 @@ void streamBackendHandleCleanup(void* arg) {
return;
}
#ifdef BUILD_NO_CALL
int32_t getLatestCheckpoint(void* arg, int64_t* checkpoint) {
SStreamMeta* pMeta = arg;
taosWLockLatch(&pMeta->chkpDirLock);
@ -738,6 +739,7 @@ int32_t delObsoleteCheckpoint(void* arg, const char* path) {
taosArrayDestroy(chkpDel);
return 0;
}
#endif
/*
* checkpointSave |--cp1--|--cp2--|--cp3--|--cp4--|--cp5--|
* chkpInUse: |--cp2--|--cp4--|
@ -855,6 +857,7 @@ int32_t streamBackendLoadCheckpointInfo(void* arg) {
return 0;
}
#ifdef BUILD_NO_CALL
int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*** ppHandle, SArray* refs) {
return 0;
// SArray* pHandle = taosArrayInit(16, POINTER_BYTES);
@ -891,6 +894,7 @@ int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*
// *ppHandle = ppCf;
// return nCf;
}
#endif
int32_t chkpGetAllDbCfHandle2(STaskDbWrapper* pBackend, rocksdb_column_family_handle_t*** ppHandle) {
SArray* pHandle = taosArrayInit(8, POINTER_BYTES);
@ -1006,6 +1010,7 @@ int32_t taskDbBuildSnap(void* arg, SArray* pSnap) {
return code;
}
#ifdef BUILD_NO_CALL
int32_t streamBackendAddInUseChkp(void* arg, int64_t chkpId) {
// if (arg == NULL) return 0;
@ -1029,6 +1034,7 @@ int32_t streamBackendDelInUseChkp(void* arg, int64_t chkpId) {
// }
// taosWUnLockLatch(&pMeta->chkpDirLock);
}
#endif
/*
0
@ -1099,7 +1105,9 @@ void streamBackendDelCompare(void* backend, void* arg) {
taosMemoryFree(node);
}
}
#ifdef BUILD_NO_CALL
void streamStateDestroy_rocksdb(SStreamState* pState, bool remove) { streamStateCloseBackend(pState, remove); }
#endif
void destroyRocksdbCfInst(RocksdbCfInst* inst) {
int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
if (inst->pHandle) {
@ -2170,6 +2178,7 @@ int32_t streamStateOpenBackendCf(void* backend, char* name, char** cfs, int32_t
taosMemoryFree(cfOpts);
return 0;
}
#ifdef BUILD_NO_CALL
int streamStateOpenBackend(void* backend, SStreamState* pState) {
taosAcquireRef(streamBackendId, pState->streamBackendRid);
SBackendWrapper* handle = backend;
@ -2279,6 +2288,7 @@ void streamStateCloseBackend(SStreamState* pState, bool remove) {
wrapper->remove |= remove; // update by other pState
taosReleaseRef(streamBackendCfWrapperId, pState->pTdbState->backendCfWrapperId);
}
#endif
void streamStateDestroyCompar(void* arg) {
SCfComparator* comp = (SCfComparator*)arg;
for (int i = 0; i < comp->numOfComp; i++) {
@ -2669,7 +2679,7 @@ SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState) {
STREAM_STATE_DEL_ROCKSDB(pState, "state", &maxStateKey);
return pCur;
}
#ifdef BUILD_NO_CALL
SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey* key) {
qDebug("streamStateGetCur_rocksdb");
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -2719,6 +2729,7 @@ int32_t streamStateFuncDel_rocksdb(SStreamState* pState, const STupleKey* key) {
STREAM_STATE_DEL_ROCKSDB(pState, "func", key);
return 0;
}
#endif
// session cf
int32_t streamStateSessionPut_rocksdb(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) {
@ -3128,6 +3139,7 @@ SStreamStateCur* streamStateFillSeekKeyPrev_rocksdb(SStreamState* pState, const
streamStateFreeCur(pCur);
return NULL;
}
#ifdef BUILD_NO_CALL
int32_t streamStateSessionGetKeyByRange_rocksdb(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
stDebug("streamStateSessionGetKeyByRange_rocksdb");
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -3185,6 +3197,7 @@ int32_t streamStateSessionGetKeyByRange_rocksdb(SStreamState* pState, const SSes
streamStateFreeCur(pCur);
return -1;
}
#endif
int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal,
int32_t* pVLen) {
@ -3317,6 +3330,7 @@ _end:
return res;
}
#ifdef BUILD_NO_CALL
// partag cf
int32_t streamStatePutParTag_rocksdb(SStreamState* pState, int64_t groupId, const void* tag, int32_t tagLen) {
int code = 0;
@ -3329,6 +3343,7 @@ int32_t streamStateGetParTag_rocksdb(SStreamState* pState, int64_t groupId, void
STREAM_STATE_GET_ROCKSDB(pState, "partag", &groupId, tagVal, tagLen);
return code;
}
#endif
// parname cfg
int32_t streamStatePutParName_rocksdb(SStreamState* pState, int64_t groupId, const char tbname[TSDB_TABLE_NAME_LEN]) {
int code = 0;
@ -3342,11 +3357,13 @@ int32_t streamStateGetParName_rocksdb(SStreamState* pState, int64_t groupId, voi
return code;
}
#ifdef BUILD_NO_CALL
int32_t streamDefaultPut_rocksdb(SStreamState* pState, const void* key, void* pVal, int32_t pVLen) {
int code = 0;
STREAM_STATE_PUT_ROCKSDB(pState, "default", key, pVal, pVLen);
return code;
}
#endif
int32_t streamDefaultGet_rocksdb(SStreamState* pState, const void* key, void** pVal, int32_t* pVLen) {
int code = 0;
STREAM_STATE_GET_ROCKSDB(pState, "default", key, pVal, pVLen);
@ -3400,6 +3417,7 @@ int32_t streamDefaultIterGet_rocksdb(SStreamState* pState, const void* start, co
rocksdb_iter_destroy(pIter);
return code;
}
#ifdef BUILD_NO_CALL
void* streamDefaultIterCreate_rocksdb(SStreamState* pState) {
SStreamStateCur* pCur = createStreamStateCursor();
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -3443,6 +3461,7 @@ char* streamDefaultIterVal_rocksdb(void* iter, int32_t* len) {
return ret;
}
#endif
// batch func
void* streamStateCreateBatch() {
rocksdb_writebatch_t* pBatch = rocksdb_writebatch_create();
@ -3796,11 +3815,12 @@ void dbChkpDestroy(SDbChkp* pChkp) {
taosMemoryFree(pChkp->pManifest);
taosMemoryFree(pChkp);
}
#ifdef BUILD_NO_CALL
int32_t dbChkpInit(SDbChkp* p) {
if (p == NULL) return 0;
return 0;
}
#endif
int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
taosThreadRwlockRdlock(&p->rwLock);
int32_t code = -1;
@ -3945,6 +3965,7 @@ int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list,
return code;
}
#ifdef BUILD_NO_CALL
int32_t bkdMgtAddChkp(SBkdMgt* bm, char* task, char* path) {
int32_t code = -1;
@ -3974,4 +3995,5 @@ int32_t bkdMgtDumpTo(SBkdMgt* bm, char* taskId, char* dname) {
taosThreadRwlockUnlock(&bm->rwLock);
return code;
}
}
#endif

2
source/libs/stream/src/streamDispatch.c
View File

@ -746,7 +746,7 @@ int32_t streamTaskSendCheckpointSourceRsp(SStreamTask* pTask) {
taosArrayClear(pTask->pReadyMsgList);
stDebug("s-task:%s level:%d source checkpoint completed msg sent to mnode", pTask->id.idStr, pTask->info.taskLevel);
} else {
stDebug("s-task:%s level:%d already send rsp to mnode", pTask->id.idStr, pTask->info.taskLevel);
stDebug("s-task:%s level:%d already send rsp checkpoint success to mnode", pTask->id.idStr, pTask->info.taskLevel);
}
taosThreadMutexUnlock(&pTask->lock);

25
source/libs/stream/src/streamExec.c
View File

@ -625,10 +625,29 @@ int32_t doStreamExecTask(SStreamTask* pTask) {
// todo other thread may change the status
// do nothing after sync executor state to storage backend, untill the vnode-level checkpoint is completed.
if (type == STREAM_INPUT__CHECKPOINT) {
// todo add lock
char* p = NULL;
streamTaskGetStatus(pTask, &p);
stDebug("s-task:%s checkpoint block received, set status:%s", pTask->id.idStr, p);
streamTaskBuildCheckpoint(pTask);
ETaskStatus s = streamTaskGetStatus(pTask, &p);
if (s == TASK_STATUS__CK) {
stDebug("s-task:%s checkpoint block received, set status:%s", pTask->id.idStr, p);
streamTaskBuildCheckpoint(pTask);
} else {
// todo refactor
int32_t code = 0;
if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
code = streamTaskSendCheckpointSourceRsp(pTask);
} else {
code = streamTaskSendCheckpointReadyMsg(pTask);
}
if (code != TSDB_CODE_SUCCESS) {
// todo: let's retry send rsp to upstream/mnode
stError("s-task:%s failed to send checkpoint rsp to upstream, checkpointId:%d, code:%s", pTask->id.idStr,
0, tstrerror(code));
}
}
return 0;
}
}

2
source/libs/stream/src/streamMeta.c
View File

@ -419,6 +419,7 @@ _err:
}
// todo refactor: the lock shoud be restricted in one function
#ifdef BUILD_NO_CALL
void streamMetaInitBackend(SStreamMeta* pMeta) {
pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId, pMeta->vgId);
if (pMeta->streamBackend == NULL) {
@ -440,6 +441,7 @@ void streamMetaInitBackend(SStreamMeta* pMeta) {
pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
streamBackendLoadCheckpointInfo(pMeta);
}
#endif
void streamMetaClear(SStreamMeta* pMeta) {
// remove all existed tasks in this vnode

37
source/libs/sync/src/syncPipeline.c
View File

@ -558,22 +558,31 @@ int32_t syncFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTe
pEntry->term);
}
SRpcMsg rpcMsg = {.code = applyCode};
syncEntry2OriginalRpc(pEntry, &rpcMsg);
int32_t code = 0;
bool retry = false;
do {
SRpcMsg rpcMsg = {.code = applyCode};
syncEntry2OriginalRpc(pEntry, &rpcMsg);
SFsmCbMeta cbMeta = {0};
cbMeta.index = pEntry->index;
cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pNode, pEntry->index);
cbMeta.isWeak = pEntry->isWeak;
cbMeta.code = applyCode;
cbMeta.state = role;
cbMeta.seqNum = pEntry->seqNum;
cbMeta.term = pEntry->term;
cbMeta.currentTerm = term;
cbMeta.flag = -1;
SFsmCbMeta cbMeta = {0};
cbMeta.index = pEntry->index;
cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pNode, pEntry->index);
cbMeta.isWeak = pEntry->isWeak;
cbMeta.code = applyCode;
cbMeta.state = role;
cbMeta.seqNum = pEntry->seqNum;
cbMeta.term = pEntry->term;
cbMeta.currentTerm = term;
cbMeta.flag = -1;
(void)syncRespMgrGetAndDel(pNode->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info);
int32_t code = pFsm->FpCommitCb(pFsm, &rpcMsg, &cbMeta);
(void)syncRespMgrGetAndDel(pNode->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info);
code = pFsm->FpCommitCb(pFsm, &rpcMsg, &cbMeta);
retry = (code != 0) && (terrno == TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE);
if (retry) {
taosMsleep(10);
sError("vgId:%d, retry on fsm commit since %s. index:%" PRId64, pNode->vgId, terrstr(), pEntry->index);
}
} while (retry);
return code;
}

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

@ -806,6 +806,13 @@ static int32_t syncNodeOnSnapshotBegin(SSyncNode *pSyncNode, SyncSnapshotSend *p
goto _SEND_REPLY;
}
SyncIndex beginIndex = syncNodeGetSnapBeginIndex(pSyncNode);
if (pReceiver->snapshotParam.start != beginIndex) {
sRError(pReceiver, "snapshot begin index is changed unexpectedly. sver:%" PRId64 ", beginIndex:%" PRId64,
pReceiver->snapshotParam.start, beginIndex);
goto _SEND_REPLY;
}
code = 0;
_SEND_REPLY:
if (code != 0 && terrno != 0) {

13
source/util/src/tcompare.c
View File

@ -1070,6 +1070,11 @@ int32_t patternMatch(const char *pattern, size_t psize, const char *str, size_t
return TSDB_PATTERN_MATCH; /* "*" at the end of the pattern matches */
}
if (c == '\\' && (pattern[i] == '_' || pattern[i] == '%')) {
c = pattern[i];
i++;
}
char rejectList[2] = {toupper(c), tolower(c)};
str += nMatchChar;
@ -1139,6 +1144,11 @@ int32_t wcsPatternMatch(const TdUcs4 *pattern, size_t psize, const TdUcs4 *str,
return TSDB_PATTERN_MATCH;
}
if (c == '\\' && (pattern[i] == '_' || pattern[i] == '%')) {
c = pattern[i];
i++;
}
TdUcs4 rejectList[2] = {towupper(c), towlower(c)};
str += nMatchChar;
@ -1166,7 +1176,8 @@ int32_t wcsPatternMatch(const TdUcs4 *pattern, size_t psize, const TdUcs4 *str,
c1 = str[j++];
nMatchChar++;
if (c == L'\\' && pattern[i] == L'_' && c1 == L'_') {
if (c == '\\' && pattern[i] == c1 &&
(c1 == '_' || c1 == '%')) {
i++;
continue;
}

56
source/util/test/utilTests.cpp
View File

@ -76,6 +76,16 @@ TEST(utilTest, wchar_pattern_match_test) {
const TdWchar* str12 = L"";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern12), 4, reinterpret_cast<const TdUcs4*>(str12), 0, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOMATCH);
const TdWchar* pattern13 = L"%\\_6 ";
const TdWchar* str13 = L"6a6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern13), 6, reinterpret_cast<const TdUcs4*>(str13), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
const TdWchar* pattern14 = L"%\\%6 ";
const TdWchar* str14 = L"6a6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern14), 6, reinterpret_cast<const TdUcs4*>(str14), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
}
TEST(utilTest, wchar_pattern_match_no_terminated) {
@ -126,14 +136,24 @@ TEST(utilTest, wchar_pattern_match_no_terminated) {
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern8), 8, reinterpret_cast<const TdUcs4*>(str8), 6, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
const TdWchar* pattern9 = L"6\\_6 ";
const TdWchar* pattern9 = L"6\\_6 ";
const TdWchar* str9 = L"6_6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern9), 4, reinterpret_cast<const TdUcs4*>(str9), 3, &pInfo);
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern9), 6, reinterpret_cast<const TdUcs4*>(str9), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const TdWchar* pattern10 = L"% ";
const TdWchar* str10 = L"6_6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern10), 1, reinterpret_cast<const TdUcs4*>(str10), 3, &pInfo);
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern10), 2, reinterpret_cast<const TdUcs4*>(str10), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const TdWchar* pattern11 = L"%\\_6 ";
const TdWchar* str11 = L"6_6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern11), 6, reinterpret_cast<const TdUcs4*>(str11), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const TdWchar* pattern12 = L"%\\%6 ";
const TdWchar* str12 = L"6%6 ";
ret = wcsPatternMatch(reinterpret_cast<const TdUcs4*>(pattern12), 6, reinterpret_cast<const TdUcs4*>(str12), 4, &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
}
@ -209,6 +229,36 @@ TEST(utilTest, char_pattern_match_test) {
const char* str13 = "a%c";
ret = patternMatch(pattern13, 5, str13, strlen(str13), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const char* pattern14 = "%a\\%c";
const char* str14 = "a%c";
ret = patternMatch(pattern14, strlen(pattern14), str14, strlen(str14), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const char* pattern15 = "_a\\%c";
const char* str15 = "ba%c";
ret = patternMatch(pattern15, strlen(pattern15), str15, strlen(str15), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const char* pattern16 = "_\\%c";
const char* str16 = "a%c";
ret = patternMatch(pattern16, strlen(pattern16), str16, strlen(str16), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_MATCH);
const char* pattern17 = "_\\%c";
const char* str17 = "ba%c";
ret = patternMatch(pattern17, strlen(pattern17), str17, strlen(str17), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOMATCH);
const char* pattern18 = "%\\%c";
const char* str18 = "abc";
ret = patternMatch(pattern18, strlen(pattern18), str18, strlen(str18), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
const char* pattern19 = "%\\_c";
const char* str19 = "abc";
ret = patternMatch(pattern19, strlen(pattern19), str19, strlen(str19), &pInfo);
ASSERT_EQ(ret, TSDB_PATTERN_NOWILDCARDMATCH);
}
TEST(utilTest, char_pattern_match_no_terminated) {

60
tests/army/community/cluster/snapshot.json Normal file
View File

@ -0,0 +1,60 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"connection_pool_size": 8,
"num_of_records_per_req": 2000,
"thread_count": 2,
"create_table_thread_count": 1,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "db",
"drop": "yes",
"vgroups": 2,
"replica": 3,
"duration":"1d",
"wal_retention_period": 1,
"wal_retention_size": 1,
"keep": "3d,6d,30d"
},
"super_tables": [
{
"name": "stb",
"child_table_exists": "no",
"childtable_count": 10,
"insert_rows": 100000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"timestamp_step": 10000,
"start_timestamp":"now-12d",
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc" },
{ "type": "double", "name": "dc"},
{ "type": "tinyint", "name": "ti"},
{ "type": "smallint", "name": "si" },
{ "type": "int", "name": "ic" },
{ "type": "bigint", "name": "bi" },
{ "type": "utinyint", "name": "uti"},
{ "type": "usmallint", "name": "usi"},
{ "type": "uint", "name": "ui" },
{ "type": "ubigint", "name": "ubi"},
{ "type": "binary", "name": "bin", "len": 16},
{ "type": "nchar", "name": "nch", "len": 32}
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"name": "location","type": "binary", "len": 16, "values":
["San Francisco", "Los Angles", "San Diego", "San Jose", "Palo Alto", "Campbell", "Mountain View","Sunnyvale", "Santa Clara", "Cupertino"]
}
]
}
]
}
]
}

96
tests/army/community/cluster/snapshot.py Normal file
View File

@ -0,0 +1,96 @@
###################################################################
# 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 -*-
import sys
import time
import random
import taos
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def insertData(self):
tdLog.info(f"insert data.")
# taosBenchmark run
jfile = etool.curFile(__file__, "snapshot.json")
etool.runBenchmark(json=jfile)
tdSql.execute(f"use {self.db}")
# set insert data information
self.childtable_count = 10
self.insert_rows = 100000
self.timestamp_step = 10000
def doAction(self):
tdLog.info(f"do action.")
self.flushDb()
# split vgroups
self.splitVGroups()
self.trimDb()
self.checkAggCorrect()
# balance vgroups
self.balanceVGroupLeader()
# replica to 1
self.alterReplica(1)
self.checkAggCorrect()
self.compactDb()
self.alterReplica(3)
vgids = self.getVGroup(self.db)
selid = random.choice(vgids)
self.balanceVGroupLeaderOn(selid)
# run
def run(self):
tdLog.debug(f"start to excute {__file__}")
# insert data
self.insertData()
# check insert data correct
self.checkInsertCorrect()
# save
self.snapshotAgg()
# do action
self.doAction()
# check save agg result correct
self.checkAggCorrect()
# check insert correct again
self.checkInsertCorrect()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

58
tests/army/enterprise/s3/s3_basic.json Normal file
View File

@ -0,0 +1,58 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"connection_pool_size": 8,
"num_of_records_per_req": 2000,
"thread_count": 2,
"create_table_thread_count": 1,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "db",
"drop": "yes",
"vgroups": 2,
"replica": 1,
"duration":"1d",
"keep": "3d,6d,30d"
},
"super_tables": [
{
"name": "stb",
"child_table_exists": "no",
"childtable_count": 4,
"insert_rows": 1000000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"timestamp_step": 1000,
"start_timestamp":"now-13d",
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc" },
{ "type": "double", "name": "dc"},
{ "type": "tinyint", "name": "ti", "values":["1"]},
{ "type": "smallint", "name": "si" },
{ "type": "int", "name": "ic" },
{ "type": "bigint", "name": "bi" },
{ "type": "utinyint", "name": "uti"},
{ "type": "usmallint", "name": "usi"},
{ "type": "uint", "name": "ui" },
{ "type": "ubigint", "name": "ubi"},
{ "type": "binary", "name": "bin", "len": 32},
{ "type": "nchar", "name": "nch", "len": 64}
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"name": "location","type": "binary", "len": 16, "values":
["San Francisco", "Los Angles", "San Diego", "San Jose", "Palo Alto", "Campbell", "Mountain View","Sunnyvale", "Santa Clara", "Cupertino"]
}
]
}
]
}
]
}

114
tests/army/enterprise/s3/s3_basic.py Normal file
View File

@ -0,0 +1,114 @@
###################################################################
# 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 -*-
import sys
import time
import taos
import frame
import frame.etool
import frame.eos
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame.srvCtl import *
from frame import *
from frame.eos import *
#
# 192.168.1.52 MINIO S3 API KEY: MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa
#
'''
s3EndPoint http://192.168.1.52:9000
s3AccessKey MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa
s3BucketName ci-bucket
s3UploadDelaySec 60
'''
class TDTestCase(TBase):
updatecfgDict = {
's3EndPoint': 'http://192.168.1.52:9000',
's3AccessKey': 'MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa',
's3BucketName': 'ci-bucket',
's3BlockSize': '10240',
's3BlockCacheSize': '320',
's3PageCacheSize': '10240',
's3UploadDelaySec':'60'
}
def insertData(self):
tdLog.info(f"insert data.")
# taosBenchmark run
json = etool.curFile(__file__, "s3_basic.json")
etool.runBenchmark(json=json)
tdSql.execute(f"use {self.db}")
# set insert data information
self.childtable_count = 4
self.insert_rows = 1000000
self.timestamp_step = 1000
def doAction(self):
tdLog.info(f"do action.")
self.flushDb()
self.compactDb()
# sleep 70s
tdLog.info(f"wait 65s ...")
time.sleep(65)
self.trimDb(True)
rootPath = sc.clusterRootPath()
cmd = f"ls {rootPath}/dnode1/data20/vnode/vnode*/tsdb/*.data"
tdLog.info(cmd)
loop = 0
while len(eos.runRetList(cmd)) > 0 and loop < 40:
time.sleep(5)
self.trimDb(True)
loop += 1
# run
def run(self):
tdLog.debug(f"start to excute {__file__}")
# insert data
self.insertData()
# check insert data correct
self.checkInsertCorrect()
# save
self.snapshotAgg()
# do action
self.doAction()
# check save agg result correct
self.checkAggCorrect()
# check insert correct again
self.checkInsertCorrect()
# drop database and free s3 file
self.dropDb()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

112
tests/army/frame/caseBase.py
View File

@ -15,6 +15,7 @@ import sys
import os
import time
import datetime
import random
from frame.log import *
from frame.sql import *
@ -42,17 +43,13 @@ class TBase:
self.db = "db"
self.stb = "stb"
# variant in taosBenchmark json
self.childtable_count = 2
self.insert_rows = 1000000
self.timestamp_step = 1000
# sql
self.sqlSum = f"select sum(ic) from {self.stb}"
self.sqlMax = f"select max(ic) from {self.stb}"
self.sqlMin = f"select min(ic) from {self.stb}"
self.sqlAvg = f"select avg(ic) from {self.stb}"
self.sqlFirst = f"select first(ts) from {self.stb}"
self.sqlLast = f"select last(ts) from {self.stb}"
# stop
def stop(self):
@ -63,14 +60,62 @@ class TBase:
# db action
#
def trimDb(self):
tdSql.execute(f"trim database {self.db}")
def trimDb(self, show = False):
tdSql.execute(f"trim database {self.db}", show = show)
def compactDb(self):
tdSql.execute(f"compact database {self.db}")
def compactDb(self, show = False):
tdSql.execute(f"compact database {self.db}", show = show)
def flushDb(self, show = False):
tdSql.execute(f"flush database {self.db}", show = show)
def dropDb(self, show = False):
tdSql.execute(f"drop database {self.db}", show = show)
def splitVGroups(self):
vgids = self.getVGroup(self.db)
selid = random.choice(vgids)
sql = f"split vgroup {selid}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def alterReplica(self, replica):
sql = f"alter database {self.db} replica {replica}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroup(self):
sql = f"balance vgroup"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroupLeader(self):
sql = f"balance vgroup leader"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroupLeaderOn(self, vgId):
sql = f"balance vgroup leader on {vgId}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def flushDb(self):
tdSql.execute(f"flush database {self.db}")
#
# check db correct
@ -91,12 +136,13 @@ class TBase:
tdSql.checkAgg(sql, 0)
# save agg result
def snapshotAgg(self):
def snapshotAgg(self):
self.sum = tdSql.getFirstValue(self.sqlSum)
self.avg = tdSql.getFirstValue(self.sqlAvg)
self.min = tdSql.getFirstValue(self.sqlMin)
self.max = tdSql.getFirstValue(self.sqlMax)
self.first = tdSql.getFirstValue(self.sqlFirst)
self.last = tdSql.getFirstValue(self.sqlLast)
# check agg
def checkAggCorrect(self):
@ -104,3 +150,41 @@ class TBase:
tdSql.checkAgg(self.sqlAvg, self.avg)
tdSql.checkAgg(self.sqlMin, self.min)
tdSql.checkAgg(self.sqlMax, self.max)
tdSql.checkAgg(self.sqlFirst, self.first)
tdSql.checkAgg(self.sqlLast, self.last)
#
# get db information
#
# get vgroups
def getVGroup(self, db_name):
vgidList = []
sql = f"select vgroup_id from information_schema.ins_vgroups where db_name='{db_name}'"
res = tdSql.getResult(sql)
rows = len(res)
for i in range(rows):
vgidList.append(res[i][0])
return vgidList
#
# util
#
# wait transactions count to zero , return False is translation not finished
def waitTransactionZero(self, seconds = 300, interval = 1):
# wait end
for i in range(seconds):
sql ="show transactions;"
rows = tdSql.query(sql)
if rows == 0:
tdLog.info("transaction count became zero.")
return True
#tdLog.info(f"i={i} wait ...")
time.sleep(interval)
return False

21
tests/army/frame/eos.py
View File

@ -20,11 +20,16 @@ import os
import time
import datetime
import platform
import subprocess
# if windows platform return True
def isWin():
return platform.system().lower() == 'windows'
#
# execute programe
#
# wait util execute file finished
def exe(file):
return os.system(file)
@ -34,3 +39,19 @@ def exeNoWait(file):
print("exe no wait")
# run return output and error
def run(command):
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
process.wait(3)
output = process.stdout.read().decode(encoding="gbk")
error = process.stderr.read().decode(encoding="gbk")
return output, error
# return list after run
def runRetList(command):
lines = []
output,error = run(command)
return output.splitlines()

3
tests/army/frame/epath.py
View File

@ -51,3 +51,6 @@ def binPath():
def binFile(filename):
return binPath() + filename

17
tests/army/frame/srvCtl.py
View File

@ -16,6 +16,8 @@ import os
import time
import datetime
from frame.server.dnodes import *
class srvCtl:
def __init__(self):
# record server information
@ -24,4 +26,19 @@ class srvCtl:
self.mLevel = 0
self.mLevelDisk = 0
#
# about path
#
# get cluster root path like /root/TDinternal/sim/
def clusterRootPath(self):
return tdDnodes.getDnodesRootDir()
# return dnode data files list
def dnodeDataFiles(self, idx):
files = []
return files
sc = srvCtl()

17
tests/parallel_test/cases.task
View File

@ -5,10 +5,17 @@
#unit-test
,,y,unit-test,bash test.sh
#army-test
#
# army-test
#
,,y,army,./pytest.sh python3 ./test.py -f enterprise/multi-level/mlevel_basic.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f enterprise/s3/s3_basic.py -L 3 -D 1
,,y,army,./pytest.sh python3 ./test.py -f community/cluster/snapshot.py -N 3 -L 3 -D 2
#system test
#
# system test
#
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/stream_basic.py
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/scalar_function.py
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_interval.py
@ -29,6 +36,7 @@
,,n,system-test,python3 ./test.py -f 8-stream/snode_restart_with_checkpoint.py -N 4
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tbname_vgroup.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/compact-col.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py -Q 3
@ -131,6 +139,10 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3 -i False
,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False
@ -514,6 +526,7 @@ e
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/json_tag.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQueryInterval.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/systable_func.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/test_ts4382.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity_1.py

2
tests/system-test/0-others/splitVGroupRep1.py
View File

@ -164,7 +164,7 @@ class TDTestCase:
self.c2Sum = None
# create database db
sql = f"create database @db_name vgroups {self.vgroups1} replica 1"
sql = f"create database @db_name vgroups {self.vgroups1} replica 1 wal_retention_period 1 wal_retention_size 1"
self.exeDouble(sql)
# create super talbe st

21
tests/system-test/0-others/test_hot_refresh_configurations.py
View File

@ -159,20 +159,21 @@ class TDTestCase:
]
}
def get_param_value_with_gdb(self, config_name, process_name):
res = subprocess.Popen("gdb -q -nx -p `pidof {}` --batch -ex 'set height 0' -ex 'p {}'".format(process_name, config_name), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
r_lines = res.stdout.read()
if r_lines:
for line in r_lines.decode().split("\n"):
if "$1 = " in line:
tdLog.debug("gdb result: {}".format(line))
return line.split(" = ")[1]
def get_param_value(self, config_name):
tdSql.query("show local variables;")
for row in tdSql.queryResult:
if config_name == row[0]:
tdLog.debug("Found variable '{}'".format(row[0]))
return row[1]
def cli_check(self, name, values, except_values=False):
if not except_values:
for v in values:
tdLog.debug("Set {} to {}".format(name, v))
tdSql.execute(f'alter local "{name} {v}";')
value = self.get_param_value(name)
tdLog.debug("Get {} value: {}".format(name, value))
assert(v == int(value))
else:
for v in values:
tdLog.debug("Set {} to {}".format(name, v))
@ -190,9 +191,7 @@ class TDTestCase:
for v in values:
dnode = random.choice(p_list)
tdSql.execute(f'alter {dnode} "{name} {v}";')
if platform.system() == "Linux" and platform.machine() == "aarch64":
continue
value = self.get_param_value_with_gdb(alias, "taosd")
value = self.get_param_value(alias)
if value:
tdLog.debug(f"value: {value}")
assert(value == str(bool(v)).lower() if is_bool else str(v))

71
tests/system-test/2-query/compact-col.py Normal file
View File

@ -0,0 +1,71 @@
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import tdDnodes
from math import inf
class TDTestCase:
def caseDescription(self):
'''
case1<shenglian zhou>: [TD-]
'''
return
def init(self, conn, logSql, replicaVer=1):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), True)
self.conn = conn
def restartTaosd(self, index=1, dbname="db"):
tdDnodes.stop(index)
tdDnodes.startWithoutSleep(index)
tdSql.execute(f"use tbname_vgroup")
def run(self):
print("running {}".format(__file__))
tdSql.execute("drop database if exists tbname_vgroup")
tdSql.execute("create database if not exists tbname_vgroup")
tdSql.execute('use tbname_vgroup')
tdSql.execute('drop database if exists dbvg')
tdSql.execute('create database dbvg vgroups 8;')
tdSql.execute('use dbvg;')
tdSql.execute('create table st(ts timestamp, f int) tags (t int);')
tdSql.execute("insert into ct1 using st tags(1) values('2021-04-19 00:00:01', 1)")
tdSql.execute("insert into ct2 using st tags(2) values('2021-04-19 00:00:02', 2)")
tdSql.execute("insert into ct3 using st tags(3) values('2021-04-19 00:00:03', 3)")
tdSql.execute("insert into ct4 using st tags(4) values('2021-04-19 00:00:04', 4)")
tdSql.execute('create table st2(ts timestamp, f int) tags (t int);')
tdSql.execute("insert into ct21 using st2 tags(1) values('2021-04-19 00:00:01', 1)")
tdSql.execute("insert into ct22 using st2 tags(2) values('2021-04-19 00:00:02', 2)")
tdSql.execute("insert into ct23 using st2 tags(3) values('2021-04-19 00:00:03', 3)")
tdSql.execute("insert into ct24 using st2 tags(4) values('2021-04-19 00:00:04', 4)")
col_names = tdSql.getColNameList("compact database tbname_vgroup")
if col_names[0] != "result":
raise Exception("first column name of compact result shall be result")
col_names = tdSql.getColNameList("show variables")
if col_names[0] != "name":
raise Exception("first column name of compact result shall be name")
tdSql.execute('drop database tbname_vgroup')
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

148
tests/system-test/2-query/like.py Normal file
View File

@ -0,0 +1,148 @@
import taos
import sys
import datetime
import inspect
from util.log import *
from util.sql import *
from util.cases import *
from util.common import tdCom
class TDTestCase:
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), False)
def initDB(self):
tdSql.execute("drop database if exists db")
tdSql.execute("create database if not exists db")
def stopTest(self):
tdSql.execute("drop database if exists db")
def like_wildcard_test(self):
tdSql.execute("create table db.t1x (ts timestamp, c1 varchar(100))")
tdSql.execute("create table db.t_1x (ts timestamp, c1 varchar(100))")
tdSql.query("select * from information_schema.ins_columns where table_name like '%1x'")
tdSql.checkRows(4)
tdSql.query("select * from information_schema.ins_columns where table_name like '%\_1x'")
tdSql.checkRows(2)
tdSql.query("insert into db.t1x values(now, 'abc'), (now+1s, 'a%c'),(now+2s, 'a_c'),(now+3s, '_c'),(now+4s, '%c')")
tdSql.query("select * from db.t1x")
tdSql.checkRows(5)
tdSql.query("select * from db.t1x where c1 like '%_c'")
tdSql.checkRows(5)
tdSql.query("select * from db.t1x where c1 like '%__c'")
tdSql.checkRows(3)
tdSql.query("select * from db.t1x where c1 like '%\_c'")
tdSql.checkRows(2)
tdSql.query("select * from db.t1x where c1 like '%\%c'")
tdSql.checkRows(2)
tdSql.query("select * from db.t1x where c1 like '_\%c'")
tdSql.checkRows(1)
tdSql.checkData(0, 1, "a%c")
tdSql.query("select * from db.t1x where c1 like '_\_c'")
tdSql.checkRows(1)
tdSql.checkData(0, 1, "a_c")
tdSql.query("select * from db.t1x where c1 like '%%_c'")
tdSql.checkRows(5)
tdSql.query("select * from db.t1x where c1 like '%_%c'")
tdSql.checkRows(5)
tdSql.query("select * from db.t1x where c1 like '__%c'")
tdSql.checkRows(3)
tdSql.query("select * from db.t1x where c1 not like '__%c'")
tdSql.checkRows(2)
def like_cnc_wildcard_test(self):
tdSql.execute("create table db.t3x (ts timestamp, c1 varchar(100))")
tdSql.execute("insert into db.t3x values(now, '我是中文'), (now+1s, '我是_中文'), (now+2s, '我是%中文'), (now+3s, '%中文'),(now+4s, '_中文')")
tdSql.query("select * from db.t3x")
tdSql.checkRows(5)
tdSql.query("select * from db.t3x where c1 like '%中文'")
tdSql.checkRows(5)
tdSql.query("select * from db.t3x where c1 like '%中_文'")
tdSql.checkRows(0)
tdSql.query("select * from db.t3x where c1 like '%\%中文'")
tdSql.checkRows(2)
tdSql.query("select * from db.t3x where c1 like '%\_中文'")
tdSql.checkRows(2)
tdSql.query("select * from db.t3x where c1 like '_中文'")
tdSql.checkRows(2)
tdSql.query("select * from db.t3x where c1 like '\_中文'")
tdSql.checkRows(1)
def like_multi_wildcard_test(self):
tdSql.execute("create table db.t4x (ts timestamp, c1 varchar(100))")
# 插入测试数据
tdSql.execute("insert into db.t4x values(now, 'abc'), (now+1s, 'a%c'),(now+2s, 'a_c'),(now+3s, '_c'),(now+4s, '%c')")
tdSql.execute("insert into db.t4x values(now+5s, '%%%c'),(now+6s, '___c'),(now+7s, '%_%c'),(now+8s, '%\\c')")
tdSql.query("select * from db.t4x where c1 like '%%%_'")
tdSql.checkRows(9)
tdSql.query("select * from db.t4x where c1 like '\%\%\%_'")
tdSql.checkRows(1)
tdSql.query("select * from db.t4x where c1 like '%\_%%'")
tdSql.checkRows(4)
tdSql.query("select * from db.t4x where c1 like '_\%\%'")
tdSql.checkRows(0)
tdSql.query("select * from db.t4x where c1 like '%abc%'")
tdSql.checkRows(1)
tdSql.query("select * from db.t4x where c1 like '_%abc%'")
tdSql.checkRows(0)
tdSql.query("select * from db.t4x where c1 like '\%%\%%'")
tdSql.checkRows(2)
tdSql.query("select * from db.t4x where c1 like '\%\_%\%%'")
tdSql.checkRows(1)
def run(self):
tdLog.printNoPrefix("==========start like_wildcard_test run ...............")
tdSql.prepare(replica = self.replicaVar)
self.initDB()
self.like_wildcard_test()
self.like_cnc_wildcard_test()
self.like_multi_wildcard_test()
tdLog.printNoPrefix("==========end like_wildcard_test run ...............")
self.stopTest()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

75
tests/system-test/2-query/test_ts4382.py Normal file
View File

@ -0,0 +1,75 @@
import random
import string
from util.log import *
from util.cases import *
from util.sql import *
from util.sqlset import *
from util import constant
from util.common import *
class TDTestCase:
"""Verify the jira TS-4382
"""
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.dbname = 'db'
self.stbname = 'st'
self.ctbname_list = ["ct1", "ct2"]
self.tag_value_list = ['{"instance":"100"}', '{"instance":"200"}']
def prepareData(self):
# db
tdSql.execute("create database {};".format(self.dbname))
tdSql.execute("use {};".format(self.dbname))
tdLog.debug("Create database %s" % self.dbname)
# super table
tdSql.execute("create table {} (ts timestamp, col1 int) tags (t1 json);".format(self.stbname))
tdLog.debug("Create super table %s" % self.stbname)
# child table
for i in range(len(self.ctbname_list)):
tdSql.execute("create table {} using {} tags('{}');".format(self.ctbname_list[i], self.stbname, self.tag_value_list[i]))
tdLog.debug("Create child table %s" % self.ctbname_list)
# insert data
tdSql.execute("insert into {} values(now, 1)(now+1s, 2)".format(self.ctbname_list[0]))
tdSql.execute("insert into {} values(now, null)(now+1s, null)".format(self.ctbname_list[1]))
def run(self):
self.prepareData()
sql_list = [
# super table query with correct tag name of json type
{
"sql": "select to_char(ts, 'yyyy-mm-dd hh24:mi:ss') as time, irate(col1) from st group by to_char(ts, 'yyyy-mm-dd hh24:mi:ss'), t1->'instance' order by time;",
"result_check": "0.0"
},
# child table query with incorrect tag name of json type
{
"sql": "select to_char(ts, 'yyyy-mm-dd hh24:mi:ss') as time, irate(col1) from ct1 group by to_char(ts, 'yyyy-mm-dd hh24:mi:ss'), t1->'name' order by time;",
"result_check": "None"
},
# child table query with null value
{
"sql": "select ts, avg(col1) from ct2 group by ts, t1->'name' order by ts;",
"result_check": "None"
}
]
for sql_dic in sql_list:
tdSql.query(sql_dic["sql"])
tdLog.debug("execute sql: %s" % sql_dic["sql"])
for item in [row[1] for row in tdSql.queryResult]:
if sql_dic["result_check"] in str(item):
tdLog.debug("Check query result of '{}' successfully".format(sql_dic["sql"]))
break
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

2
tests/system-test/8-stream/max_delay_session.py
View File

@ -27,6 +27,8 @@ class TDTestCase:
self.tb_stream_des_table = f'{self.tb_name}{self.tdCom.des_table_suffix}'
self.tdCom.date_time = self.tdCom.dataDict["start_ts"]
time.sleep(1)
if watermark is not None:
watermark_value = f'{self.tdCom.dataDict["watermark"]}s'
else:

4
tests/system-test/win-test-file
View File

@ -99,6 +99,10 @@ python3 ./test.py -f 2-query/ts-4233.py
python3 ./test.py -f 2-query/ts-4233.py -Q 2
python3 ./test.py -f 2-query/ts-4233.py -Q 3
python3 ./test.py -f 2-query/ts-4233.py -Q 4
python3 ./test.py -f 2-query/like.py
python3 ./test.py -f 2-query/like.py -Q 2
python3 ./test.py -f 2-query/like.py -Q 3
python3 ./test.py -f 2-query/like.py -Q 4
python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3 -i False
python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False
python3 ./test.py -f 3-enterprise/restore/restoreMnode.py -N 5 -M 3 -i False

1
tools/shell/inc/shellInt.h
View File

@ -66,6 +66,7 @@ typedef struct {
char file[PATH_MAX];
char password[TSDB_USET_PASSWORD_LEN];
bool is_gen_auth;
bool is_bi_mode;
bool is_raw_time;
bool is_version;
bool is_dump_config;

6
tools/shell/src/shellArguments.c
View File

@ -44,6 +44,7 @@
#define SHELL_NET_ROLE "Net role when network connectivity test, options: client|server."
#define SHELL_PKT_LEN "Packet length used for net test, default is 1024 bytes."
#define SHELL_PKT_NUM "Packet numbers used for net test, default is 100."
#define SHELL_BI_MODE "Set BI mode"
#define SHELL_VERSION "Print program version."
#ifdef WEBSOCKET
@ -59,6 +60,7 @@ void shellPrintHelp() {
printf("Usage: taos [OPTION...] \r\n\r\n");
printf("%s%s%s%s\r\n", indent, "-a,", indent, SHELL_AUTH);
printf("%s%s%s%s\r\n", indent, "-A,", indent, SHELL_GEN_AUTH);
printf("%s%s%s%s\r\n", indent, "-B,", indent, SHELL_BI_MODE);
printf("%s%s%s%s\r\n", indent, "-c,", indent, SHELL_CFG_DIR);
printf("%s%s%s%s\r\n", indent, "-C,", indent, SHELL_DMP_CFG);
printf("%s%s%s%s\r\n", indent, "-d,", indent, SHELL_DB);
@ -127,6 +129,7 @@ static struct argp_option shellOptions[] = {
{"timeout", 'T', "SECONDS", 0, SHELL_TIMEOUT},
#endif
{"pktnum", 'N', "PKTNUM", 0, SHELL_PKT_NUM},
{"bimode", 'B', 0, 0, SHELL_BI_MODE},
{0},
};
@ -173,6 +176,9 @@ static int32_t shellParseSingleOpt(int32_t key, char *arg) {
case 'A':
pArgs->is_gen_auth = true;
break;
case 'B':
pArgs->is_bi_mode = true;
break;
case 'c':
#ifdef WEBSOCKET
pArgs->cloud = false;

8
tools/shell/src/shellAuto.c
View File

@ -84,8 +84,8 @@ SWords shellCommands[] = {
{"alter topic", 0, 0, NULL},
{"alter user <user_name> <user_actions> <anyword> ;", 0, 0, NULL},
#ifdef TD_ENTERPRISE
{"balance vgroup;", 0, 0, NULL},
{"balance vgroup leader <vgroup_id>", 0, 0, NULL},
{"balance vgroup ;", 0, 0, NULL},
{"balance vgroup leader on <vgroup_id>", 0, 0, NULL},
#endif
// 20
@ -531,8 +531,8 @@ void showHelp() {
printf(
"\n\n\
----- special commands on enterpise version ----- \n\
balance vgroup; \n\
balance vgroup leader <vgroup_id> \n\
balance vgroup ;\n\
balance vgroup leader on <vgroup_id> \n\
compact database <db_name>; \n\
redistribute vgroup <vgroup_id> dnode <dnode_id> ;\n\
split vgroup <vgroup_id>;");

14
tools/shell/src/shellCommand.c
View File

@ -62,7 +62,6 @@ int32_t shellCountPrefixOnes(uint8_t c) {
}
void shellGetPrevCharSize(const char *str, int32_t pos, int32_t *size, int32_t *width) {
ASSERT(pos > 0);
if (pos <= 0) return;
TdWchar wc;
@ -82,7 +81,6 @@ void shellGetPrevCharSize(const char *str, int32_t pos, int32_t *size, int32_t *
}
void shellGetNextCharSize(const char *str, int32_t pos, int32_t *size, int32_t *width) {
ASSERT(pos >= 0);
if(pos < 0) return;
TdWchar wc;
@ -91,7 +89,6 @@ void shellGetNextCharSize(const char *str, int32_t pos, int32_t *size, int32_t *
}
void shellInsertChar(SShellCmd *cmd, char *c, int32_t size) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
TdWchar wc;
@ -138,7 +135,6 @@ void shellInsertStr(SShellCmd *cmd, char *str, int32_t size) {
}
void shellBackspaceChar(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -159,7 +155,6 @@ void shellBackspaceChar(SShellCmd *cmd) {
}
void shellClearLineBefore(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);
@ -174,7 +169,6 @@ void shellClearLineBefore(SShellCmd *cmd) {
}
void shellClearLineAfter(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);
@ -184,7 +178,6 @@ void shellClearLineAfter(SShellCmd *cmd) {
}
void shellDeleteChar(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -203,7 +196,6 @@ void shellDeleteChar(SShellCmd *cmd) {
}
void shellMoveCursorLeft(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -218,7 +210,6 @@ void shellMoveCursorLeft(SShellCmd *cmd) {
}
void shellMoveCursorRight(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -233,7 +224,6 @@ void shellMoveCursorRight(SShellCmd *cmd) {
}
void shellPositionCursorHome(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -254,7 +244,6 @@ void positionCursorMiddle(SShellCmd *cmd) {
}
void shellPositionCursorEnd(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -290,7 +279,6 @@ void shellPositionCursor(int32_t step, int32_t direction) {
}
void shellUpdateBuffer(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (shellRegexMatch(cmd->buffer, "(\\s+$)|(^$)", REG_EXTENDED)) strcat(cmd->command, " ");
@ -306,7 +294,6 @@ void shellUpdateBuffer(SShellCmd *cmd) {
}
bool shellIsReadyGo(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return false;
char *total = (char *)taosMemoryCalloc(1, SHELL_MAX_COMMAND_SIZE);
@ -334,7 +321,6 @@ void shellGetMbSizeInfo(const char *str, int32_t *size, int32_t *width) {
}
void shellResetCommand(SShellCmd *cmd, const char s[]) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);

10
tools/shell/src/shellEngine.c
View File

@ -1291,6 +1291,16 @@ int32_t shellExecute() {
shellSetConn(shell.conn, runOnce);
shellReadHistory();
if(shell.args.is_bi_mode) {
// need set bi mode
printf("Set BI mode is true.\n");
#ifdef WEBSOCKET
//ws_taos_set_conn_mode(shell.ws_conn, TAOS_CONN_MODE_BI, 1);
#else
taos_set_conn_mode(shell.conn, TAOS_CONN_MODE_BI, 1);
#endif
}
if (runOnce) {
if (pArgs->commands != NULL) {
printf("%s%s\r\n", shell.info.promptHeader, pArgs->commands);