Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 2b2059ef 编写于 作者: H Haojun Liao 提交者： GitHub
浏览文件
操作

Merge pull request #20534 from taosdata/fix/TD-22671 

fix:error in TD-23218 & remove useless logic
上级 13991b61 ed53b982
展开全部 隐藏空白更改
内联 并排
Showing with 440 addition and 984 deletion
include/common/tcommon.h
浏览文件 @ 2b2059ef
......@@ -208,15 +208,12 @@ typedef struct SSDataBlock {
} SSDataBlock;
enum {
FETCH_TYPE__DATA = 1,
FETCH_TYPE__META,
FETCH_TYPE__SEP,
FETCH_TYPE__DATA = 0,
FETCH_TYPE__NONE,
};
typedef struct {
int8_t fetchType;
STqOffsetVal offset;
union {
SSDataBlock data;
void* meta;
......
include/libs/executor/executor.h
浏览文件 @ 2b2059ef
......@@ -198,11 +198,11 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
//
int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
void qStreamSetOpen(qTaskInfo_t tinfo);
SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo);
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
int64_t qStreamExtractPrepareUid(qTaskInfo_t tinfo);
SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo);
const SSchemaWrapper* qExtractSchemaFromTask(qTaskInfo_t tinfo);
......
include/libs/wal/wal.h
浏览文件 @ 2b2059ef
......@@ -146,8 +146,8 @@ typedef struct {
int64_t curFileFirstVer;
int64_t curVersion;
int64_t capacity;
int8_t curInvalid;
int8_t curStopped;
// int8_t curInvalid;
// int8_t curStopped;
TdThreadMutex mutex;
SWalFilterCond cond;
// TODO remove it
......
include/util/taoserror.h
浏览文件 @ 2b2059ef
......@@ -758,6 +758,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TMQ_INVALID_MSG TAOS_DEF_ERROR_CODE(0, 0x4000)
#define TSDB_CODE_TMQ_CONSUMER_MISMATCH TAOS_DEF_ERROR_CODE(0, 0x4001)
#define TSDB_CODE_TMQ_CONSUMER_CLOSED TAOS_DEF_ERROR_CODE(0, 0x4002)
#define TSDB_CODE_TMQ_CONSUMER_ERROR TAOS_DEF_ERROR_CODE(0, 0x4003)
// stream
#define TSDB_CODE_STREAM_TASK_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x4100)
......
source/client/src/clientSml.c
浏览文件 @ 2b2059ef
......@@ -533,8 +533,7 @@ static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSm
uint16_t *index = colHash ? (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen) : NULL;
if (index) {
if (colField[*index].type != kv->type) {
uError("SML:0x%" PRIx64 " point type and db type mismatch. key: %s. point type: %d, db type: %d", info->id,
kv->key, colField[*index].type, kv->type);
uError("SML:0x%" PRIx64 " point type and db type mismatch. point type: %d, db type: %d, key: %s", info->id, colField[*index].type, kv->type, kv->key);
return TSDB_CODE_TSC_INVALID_VALUE;
}
......
source/client/src/clientTmq.c
浏览文件 @ 2b2059ef
......@@ -1264,6 +1264,8 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
pRspWrapper->tmqRspType = TMQ_MSG_TYPE__END_RSP;
taosWriteQitem(tmq->mqueue, pRspWrapper);
}else if(code == TSDB_CODE_WAL_LOG_NOT_EXIST){ //poll data while insert
taosMsleep(500);
}
goto CREATE_MSG_FAIL;
......
source/common/src/tmsg.c
浏览文件 @ 2b2059ef
......@@ -7212,6 +7212,7 @@ void tDestroySSubmitReq2(SSubmitReq2 *pReq, int32_t flag) {
tDestroySSubmitTbData(&aSubmitTbData[i], flag);
}
taosArrayDestroy(pReq->aSubmitTbData);
pReq->aSubmitTbData = NULL;
}
int32_t tEncodeSSubmitRsp2(SEncoder *pCoder, const SSubmitRsp2 *pRsp) {
......
source/dnode/mnode/impl/inc/mndConsumer.h
浏览文件 @ 2b2059ef
......@@ -24,10 +24,10 @@ extern "C" {
enum {
MQ_CONSUMER_STATUS__MODIFY = 1,
MQ_CONSUMER_STATUS__MODIFY_IN_REB, // this value is not used anymore
// MQ_CONSUMER_STATUS__MODIFY_IN_REB, // this value is not used anymore
MQ_CONSUMER_STATUS__READY,
MQ_CONSUMER_STATUS__LOST,
MQ_CONSUMER_STATUS__LOST_IN_REB, // this value is not used anymore
// MQ_CONSUMER_STATUS__LOST_IN_REB, // this value is not used anymore
MQ_CONSUMER_STATUS__LOST_REBD,
MQ_CONSUMER_STATUS__REMOVED,
};
......
source/dnode/mnode/impl/src/mndConsumer.c
浏览文件 @ 2b2059ef
......@@ -337,7 +337,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
}
taosRUnLockLatch(&pConsumer->lock);
} else if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
} else if (status == MQ_CONSUMER_STATUS__MODIFY) {
taosRLockLatch(&pConsumer->lock);
int32_t newTopicNum = taosArrayGetSize(pConsumer->rebNewTopics);
......@@ -876,17 +876,11 @@ static void updateConsumerStatus(SMqConsumerObj* pConsumer) {
int32_t status = pConsumer->status;
if (taosArrayGetSize(pConsumer->rebNewTopics) == 0 && taosArrayGetSize(pConsumer->rebRemovedTopics) == 0) {
if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
if (status == MQ_CONSUMER_STATUS__MODIFY) {
pConsumer->status = MQ_CONSUMER_STATUS__READY;
} else if (status == MQ_CONSUMER_STATUS__LOST_IN_REB || status == MQ_CONSUMER_STATUS__LOST) {
} else if (status == MQ_CONSUMER_STATUS__LOST) {
pConsumer->status = MQ_CONSUMER_STATUS__LOST_REBD;
}
} else {
if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
pConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
} else if (status == MQ_CONSUMER_STATUS__LOST || status == MQ_CONSUMER_STATUS__LOST_IN_REB) {
pConsumer->status = MQ_CONSUMER_STATUS__LOST;
}
}
}
......@@ -1195,10 +1189,8 @@ static const char *mndConsumerStatusName(int status) {
return "ready";
case MQ_CONSUMER_STATUS__LOST:
case MQ_CONSUMER_STATUS__LOST_REBD:
case MQ_CONSUMER_STATUS__LOST_IN_REB:
return "lost";
case MQ_CONSUMER_STATUS__MODIFY:
case MQ_CONSUMER_STATUS__MODIFY_IN_REB:
return "rebalancing";
default:
return "unknown";
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 2b2059ef
......@@ -228,19 +228,12 @@ typedef struct SSnapContext {
SArray *idList;
int32_t index;
bool withMeta;
bool queryMetaOrData; // true-get meta, false-get data
bool queryMeta; // true-get meta, false-get data
} SSnapContext;
typedef struct STqReader {
// const SSubmitReq *pMsg;
// SSubmitBlk *pBlock;
// SSubmitMsgIter msgIter;
// SSubmitBlkIter blkIter;
int64_t ver;
SPackedData msg2;
int8_t setMsg;
SSubmitReq2 submit;
int32_t nextBlk;
......@@ -267,7 +260,7 @@ int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id);
int32_t tqNextBlock(STqReader *pReader, SFetchRet *ret);
void tqNextBlock(STqReader *pReader, SFetchRet *ret);
int32_t tqReaderSetSubmitReq2(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver);
// int32_t tqReaderSetDataMsg(STqReader *pReader, const SSubmitReq *pMsg, int64_t ver);
......
source/dnode/vnode/src/meta/metaSnapshot.c
浏览文件 @ 2b2059ef
......@@ -268,7 +268,7 @@ int32_t buildSnapContext(SMeta* pMeta, int64_t snapVersion, int64_t suid, int8_t
ctx->snapVersion = snapVersion;
ctx->suid = suid;
ctx->subType = subType;
ctx->queryMetaOrData = withMeta;
ctx->queryMeta = withMeta;
ctx->withMeta = withMeta;
ctx->idVersion = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
if (ctx->idVersion == NULL) {
......@@ -475,7 +475,7 @@ int32_t getMetafromSnapShot(SSnapContext* ctx, void** pBuf, int32_t* contLen, in
if (ctx->index >= taosArrayGetSize(ctx->idList)) {
metaDebug("tmqsnap get meta end");
ctx->index = 0;
ctx->queryMetaOrData = false; // change to get data
ctx->queryMeta = false; // change to get data
return 0;
}
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 2b2059ef
......@@ -236,21 +236,6 @@ int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry) {
}
int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp, int32_t type) {
#if 0
A(taosArrayGetSize(pRsp->blockData) == pRsp->blockNum);
A(taosArrayGetSize(pRsp->blockDataLen) == pRsp->blockNum);
A(!pRsp->withSchema);
A(taosArrayGetSize(pRsp->blockSchema) == 0);
if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
if (pRsp->blockNum > 0) {
A(pRsp->rspOffset.version > pRsp->reqOffset.version);
} else {
A(pRsp->rspOffset.version >= pRsp->reqOffset.version);
}
}
#endif
doSendDataRsp(&pMsg->info, pRsp, pReq->epoch, pReq->consumerId, type);
char buf1[80] = {0};
......@@ -352,17 +337,6 @@ static int32_t tqInitDataRsp(SMqDataRsp* pRsp, const SMqPollReq* pReq, int8_t su
}
pRsp->withTbName = 0;
#if 0
pRsp->withTbName = pReq->withTbName;
if (pRsp->withTbName) {
pRsp->blockTbName = taosArrayInit(0, sizeof(void*));
if (pRsp->blockTbName == NULL) {
// TODO free
return -1;
}
}
#endif
pRsp->withSchema = false;
return 0;
}
......@@ -421,6 +395,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
return -1;
}
// offset set to previous version when init
tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1);
}
} else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
......@@ -461,7 +436,6 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
SRpcMsg* pMsg, STqOffsetVal* pOffset) {
int32_t code = 0;
uint64_t consumerId = pRequest->consumerId;
int32_t vgId = TD_VID(pTq->pVnode);
......@@ -472,10 +446,9 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
taosWLockLatch(&pTq->lock);
qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
code = tqScanData(pTq, pHandle, &dataRsp, pOffset);
if (code != TSDB_CODE_SUCCESS) {
taosWUnLockLatch(&pTq->lock);
return code;
int code = tqScanData(pTq, pHandle, &dataRsp, pOffset);
if(code != 0) {
goto end;
}
// till now, all data has been transferred to consumer, new data needs to push client once arrived.
......@@ -486,89 +459,60 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
return code;
}
taosWUnLockLatch(&pTq->lock);
code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&dataRsp, TMQ_MSG_TYPE__POLL_RSP);
// NOTE: this pHandle->consumerId may have been changed already.
tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, offset type:%d, uid/version:%" PRId64
", ts:%" PRId64 ", reqId:0x%" PRIx64,
consumerId, pHandle->subKey, vgId, dataRsp.blockNum, dataRsp.rspOffset.type, dataRsp.rspOffset.uid,
dataRsp.rspOffset.ts, pRequest->reqId);
tDeleteSMqDataRsp(&dataRsp);
end:
{
char buf[80] = {0};
tFormatOffset(buf, 80, &dataRsp.rspOffset);
tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, rsp offset type:%s, reqId:0x%" PRIx64 " code:%d",
consumerId, pHandle->subKey, vgId, dataRsp.blockNum, buf, pRequest->reqId, code);
taosWUnLockLatch(&pTq->lock);
tDeleteSMqDataRsp(&dataRsp);
}
return code;
}
static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
int32_t code = -1;
STqOffsetVal offset = {0};
SWalCkHead* pCkHead = NULL;
int32_t vgId = TD_VID(pTq->pVnode);
STqOffsetVal reqOffset = pRequest->reqOffset;
uint64_t consumerId = pRequest->consumerId;
// 1. reset the offset if needed
if (IS_OFFSET_RESET_TYPE(reqOffset.type)) {
// handle the reset offset cases, according to the consumer's choice.
bool blockReturned = false;
code = extractResetOffsetVal(&offset, pTq, pHandle, pRequest, pMsg, &blockReturned);
if (code != 0) {
return code;
}
// empty block returned, quit
if (blockReturned) {
return 0;
}
} else { // use the consumer specified offset
// the offset value can not be monotonious increase??
offset = reqOffset;
}
// this is a normal subscribe requirement
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
return extractDataAndRspForNormalSubscribe(pTq, pHandle, pRequest, pMsg, &offset);
}
// todo handle the case where re-balance occurs.
// for taosx
static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg, STqOffsetVal *offset) {
int code = 0;
int32_t vgId = TD_VID(pTq->pVnode);
SWalCkHead* pCkHead = NULL;
SMqMetaRsp metaRsp = {0};
STaosxRsp taosxRsp = {0};
tqInitTaosxRsp(&taosxRsp, pRequest);
if (offset.type != TMQ_OFFSET__LOG) {
if (tqScanTaosx(pTq, pHandle, &taosxRsp, &metaRsp, &offset) < 0) {
if (offset->type != TMQ_OFFSET__LOG) {
if (tqScanTaosx(pTq, pHandle, &taosxRsp, &metaRsp, offset) < 0) {
return -1;
}
if (metaRsp.metaRspLen > 0) {
code = tqSendMetaPollRsp(pTq, pMsg, pRequest, &metaRsp);
tqDebug("tmq poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send meta offset type:%d,uid:%" PRId64
",ts:%" PRId64,
consumerId, pHandle->subKey, vgId, metaRsp.rspOffset.type, metaRsp.rspOffset.uid,
metaRsp.rspOffset.ts);
tqDebug("tmq poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send meta offset type:%d,uid:%" PRId64 ",ts:%" PRId64,
pRequest->consumerId, pHandle->subKey, vgId, metaRsp.rspOffset.type, metaRsp.rspOffset.uid, metaRsp.rspOffset.ts);
taosMemoryFree(metaRsp.metaRsp);
tDeleteSTaosxRsp(&taosxRsp);
return code;
}
tqDebug("taosx poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send data blockNum:%d, offset type:%d,uid:%" PRId64
",ts:%" PRId64,pRequest->consumerId, pHandle->subKey, vgId, taosxRsp.blockNum, taosxRsp.rspOffset.type, taosxRsp.rspOffset.uid,taosxRsp.rspOffset.ts);
if (taosxRsp.blockNum > 0) {
code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP);
tDeleteSTaosxRsp(&taosxRsp);
return code;
} else {
offset = taosxRsp.rspOffset;
}else {
*offset = taosxRsp.rspOffset;
}
tqDebug("taosx poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send data blockNum:%d, offset type:%d,uid:%" PRId64
",version:%" PRId64,
consumerId, pHandle->subKey, vgId, taosxRsp.blockNum, taosxRsp.rspOffset.type, taosxRsp.rspOffset.uid,
taosxRsp.rspOffset.version);
}
if (offset.type == TMQ_OFFSET__LOG) {
int64_t fetchVer = offset.version + 1;
if (offset->type == TMQ_OFFSET__LOG) {
int64_t fetchVer = offset->version + 1;
pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048);
if (pCkHead == NULL) {
tDeleteSTaosxRsp(&taosxRsp);
......@@ -578,12 +522,10 @@ static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* p
walSetReaderCapacity(pHandle->pWalReader, 2048);
int totalRows = 0;
while (1) {
// todo refactor: this is not correct.
int32_t savedEpoch = atomic_load_32(&pHandle->epoch);
if (savedEpoch > pRequest->epoch) {
tqWarn("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey:%s vgId:%d offset %" PRId64
", found new consumer epoch %d, discard req epoch %d",
consumerId, pRequest->epoch, pHandle->subKey, vgId, fetchVer, savedEpoch, pRequest->epoch);
", found new consumer epoch %d, discard req epoch %d", pRequest->consumerId, pRequest->epoch, pHandle->subKey, vgId, fetchVer, savedEpoch, pRequest->epoch);
break;
}
......@@ -596,7 +538,7 @@ static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* p
}
SWalCont* pHead = &pCkHead->head;
tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, vgId:%d offset %" PRId64 " msgType %d", consumerId,
tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, vgId:%d offset %" PRId64 " msgType %d", pRequest->consumerId,
pRequest->epoch, vgId, fetchVer, pHead->msgType);
// process meta
......@@ -634,7 +576,7 @@ static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* p
};
if (tqTaosxScanLog(pTq, pHandle, submit, &taosxRsp, &totalRows) < 0) {
tqError("tmq poll: tqTaosxScanLog error %" PRId64 ", in vgId:%d, subkey %s", consumerId, vgId,
tqError("tmq poll: tqTaosxScanLog error %" PRId64 ", in vgId:%d, subkey %s", pRequest->consumerId, vgId,
pRequest->subKey);
taosMemoryFreeClear(pCkHead);
tDeleteSTaosxRsp(&taosxRsp);
......@@ -658,6 +600,39 @@ static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* p
return 0;
}
static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
int32_t code = -1;
STqOffsetVal offset = {0};
STqOffsetVal reqOffset = pRequest->reqOffset;
// 1. reset the offset if needed
if (IS_OFFSET_RESET_TYPE(reqOffset.type)) {
// handle the reset offset cases, according to the consumer's choice.
bool blockReturned = false;
code = extractResetOffsetVal(&offset, pTq, pHandle, pRequest, pMsg, &blockReturned);
if (code != 0) {
return code;
}
// empty block returned, quit
if (blockReturned) {
return 0;
}
} else { // use the consumer specified offset
// the offset value can not be monotonious increase??
offset = reqOffset;
}
// this is a normal subscribe requirement
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
return extractDataAndRspForNormalSubscribe(pTq, pHandle, pRequest, pMsg, &offset);
}
// todo handle the case where re-balance occurs.
// for taosx
return extractDataAndRspForDbStbSubscribe(pTq, pHandle, pRequest, pMsg, &offset);
}
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
SMqPollReq req = {0};
if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
......
source/dnode/vnode/src/tq/tqPush.c
浏览文件 @ 2b2059ef
......@@ -238,7 +238,7 @@ static void doPushDataForEntry(void* pIter, STqExecHandle* pExec, STQ* pTq, int6
if (qStreamSetScanMemData(pTaskInfo, submit) != 0) {
return;
}
qStreamSetOpen(pTaskInfo);
// here start to scan submit block to extract the subscribed data
int32_t totalRows = 0;
......
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 2b2059ef
此差异已折叠。
source/dnode/vnode/src/tq/tqScan.c
浏览文件 @ 2b2059ef
......@@ -71,37 +71,25 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
int32_t totalRows = 0;
const STqExecHandle* pExec = &pHandle->execHandle;
qTaskInfo_t task = pExec->task;
qTaskInfo_t task = pExec->task;
if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
if (pOffset->type == TMQ_OFFSET__LOG) {
pRsp->rspOffset = *pOffset;
return code;
} else {
tqOffsetResetToLog(pOffset, pHandle->snapshotVer);
if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
tqDebug("prepare scan failed, vgId:%d, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
pRsp->rspOffset = *pOffset;
return code;
}
}
tqError("prepare scan failed, return");
return -1;
}
while (1) {
SSDataBlock* pDataBlock = NULL;
uint64_t ts = 0;
tqDebug("vgId:%d, tmq task start to execute, consumer:0x%"PRIx64, vgId, pHandle->consumerId);
code = qExecTask(task, &pDataBlock, &ts);
if (code != TSDB_CODE_SUCCESS) {
tqError("vgId:%d, task exec error since %s, consumer:0x%" PRIx64, vgId, terrstr(),
pHandle->consumerId);
return code;
qStreamSetOpen(task);
tqDebug("consumer:0x%"PRIx64" vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);
if (qExecTask(task, &pDataBlock, &ts) != TSDB_CODE_SUCCESS) {
tqError("consumer:0x%"PRIx64" vgId:%d, task exec error since %s", pHandle->consumerId, vgId, terrstr());
return -1;
}
// current scan should be stopped ASAP, since the re-balance occurs.
tqDebug("consumer:0x%"PRIx64" vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, pDataBlock);
// current scan should be stopped asap, since the rebalance occurs.
if (pDataBlock == NULL) {
break;
}
......@@ -113,37 +101,15 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
}
pRsp->blockNum++;
tqDebug("vgId:%d, consumer:0x%" PRIx64 " tmq task executed, rows:%" PRId64 ", total blocks:%d", vgId, pHandle->consumerId,
pDataBlock->info.rows, pRsp->blockNum);
if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
totalRows += pDataBlock->info.rows;
if (totalRows >= MAX_ROWS_TO_RETURN) {
break;
}
totalRows += pDataBlock->info.rows;
if (totalRows >= MAX_ROWS_TO_RETURN) {
break;
}
}
tqDebug("consumer:0x%"PRIx64" vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);
qStreamExtractOffset(task, &pRsp->rspOffset);
if (pRsp->rspOffset.type == 0) {
code = TSDB_CODE_INVALID_PARA;
tqError("vgId:%d, expected rsp offset: type %d %" PRId64 " %" PRId64 " %" PRId64, vgId, pRsp->rspOffset.type,
pRsp->rspOffset.ts, pRsp->rspOffset.uid, pRsp->rspOffset.version);
return code;
}
if (pRsp->withTbName || pRsp->withSchema) {
code = TSDB_CODE_INVALID_PARA;
tqError("vgId:%d, get column should not with meta:%d,%d", vgId, pRsp->withTbName, pRsp->withSchema);
return code;
}
tqDebug("vgId:%d, consumer:0x%" PRIx64 " tmq task executed, total blocks:%d, rows:%d", vgId, pHandle->consumerId,
pRsp->blockNum, totalRows);
return code;
return 0;
}
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
......@@ -151,18 +117,8 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
qTaskInfo_t task = pExec->task;
if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
tqDebug("prepare scan failed, return");
if (pOffset->type == TMQ_OFFSET__LOG) {
pRsp->rspOffset = *pOffset;
return 0;
} else {
tqOffsetResetToLog(pOffset, pHandle->snapshotVer);
if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
tqDebug("prepare scan failed, return");
pRsp->rspOffset = *pOffset;
return 0;
}
}
tqDebug("tqScanTaosx prepare scan failed, return");
return -1;
}
int32_t rowCnt = 0;
......@@ -208,42 +164,31 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
}
}
if (pDataBlock == NULL && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
if (qStreamExtractPrepareUid(task) != 0) {
// get meta
SMqMetaRsp* tmp = qStreamExtractMetaMsg(task);
if (tmp->metaRspLen > 0) {
qStreamExtractOffset(task, &tmp->rspOffset);
*pMetaRsp = *tmp;
tqDebug("tmqsnap task get data");
break;
}
if (pDataBlock == NULL) {
qStreamExtractOffset(task, pOffset);
if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
continue;
}
tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode),
pHandle->snapshotVer + 1);
tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);
qStreamExtractOffset(task, &pRsp->rspOffset);
break;
}
if (pRsp->blockNum > 0) {
tqDebug("tmqsnap task exec exited, get data");
qStreamExtractOffset(task, &pRsp->rspOffset);
break;
}
SMqMetaRsp* tmp = qStreamExtractMetaMsg(task);
if (tmp->rspOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
tqOffsetResetToData(pOffset, tmp->rspOffset.uid, tmp->rspOffset.ts);
qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);
tmp->rspOffset.type = TMQ_OFFSET__SNAPSHOT_META;
tqDebug("tmqsnap task exec change to get data");
continue;
}
*pMetaRsp = *tmp;
tqDebug("tmqsnap task exec exited, get meta");
tqDebug("task exec exited");
break;
}
qStreamExtractOffset(task, &pRsp->rspOffset);
if (pRsp->rspOffset.type == 0) {
tqError("expected rsp offset: type %d %" PRId64 " %" PRId64 " %" PRId64, pRsp->rspOffset.type, pRsp->rspOffset.ts,
pRsp->rspOffset.uid, pRsp->rspOffset.version);
return -1;
}
return 0;
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 2b2059ef
......@@ -128,10 +128,10 @@ enum {
typedef struct {
// TODO remove prepareStatus
STqOffsetVal prepareStatus; // for tmq
STqOffsetVal lastStatus; // for tmq
// STqOffsetVal prepareStatus; // for tmq
STqOffsetVal currentOffset; // for tmq
SMqMetaRsp metaRsp; // for tmq fetching meta
int8_t returned;
// int8_t returned;
int64_t snapshotVer;
// const SSubmitReq* pReq;
......@@ -191,7 +191,7 @@ enum {
OP_OPENED = 0x1,
OP_RES_TO_RETURN = 0x5,
OP_EXEC_DONE = 0x9,
OP_EXEC_RECV = 0x11,
// OP_EXEC_RECV = 0x11,
};
typedef struct SOperatorFpSet {
......
source/libs/executor/src/executor.c
浏览文件 @ 2b2059ef
......@@ -1027,15 +1027,9 @@ SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo) {
return &pTaskInfo->streamInfo.metaRsp;
}
int64_t qStreamExtractPrepareUid(qTaskInfo_t tinfo) {
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
return pTaskInfo->streamInfo.prepareStatus.uid;
}
int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
memcpy(pOffset, &pTaskInfo->streamInfo.lastStatus, sizeof(STqOffsetVal));
return 0;
memcpy(pOffset, &pTaskInfo->streamInfo.currentOffset, sizeof(STqOffsetVal));
}
int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* sContext, SMetaTableInfo* pMtInfo) {
......@@ -1081,22 +1075,27 @@ int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit) {
return 0;
}
void qStreamSetOpen(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot;
pOperator->status = OP_NOT_OPENED;
}
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot;
const char* id = GET_TASKID(pTaskInfo);
pTaskInfo->streamInfo.prepareStatus = *pOffset;
pTaskInfo->streamInfo.returned = 0;
if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus)) {
// if pOffset equal to current offset, means continue consume
if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) {
return 0;
}
if (subType == TOPIC_SUB_TYPE__COLUMN) {
pOperator->status = OP_OPENED;
pOperator = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id);
if (pOperator == NULL) {
return -1;
}
SStreamScanInfo* pInfo = pOperator->info;
STableScanInfo* pScanInfo = pInfo->pTableScanOp->info;
STableScanBase* pScanBaseInfo = &pScanInfo->base;
......@@ -1106,7 +1105,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
tsdbReaderClose(pScanBaseInfo->dataReader);
pScanBaseInfo->dataReader = NULL;
// let's seek to the next version in wal file
// set version to read for wal is next, so +1
if (tqSeekVer(pInfo->tqReader, pOffset->version + 1, pTaskInfo->id.str) < 0) {
qError("tqSeekVer failed ver:%"PRId64", %s", pOffset->version + 1, id);
return -1;
......@@ -1135,6 +1134,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
}
}
qDebug("switch to table uid:%" PRId64 " ts:%" PRId64 "% "PRId64 " rows returned", uid, ts, pInfo->pTableScanOp->resultInfo.totalRows);
pInfo->pTableScanOp->resultInfo.totalRows = 0;
// start from current accessed position
......@@ -1239,6 +1239,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id);
}
}
pTaskInfo->streamInfo.currentOffset = *pOffset;
return 0;
}
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 2b2059ef
......@@ -1967,11 +1967,11 @@ void qStreamCloseTsdbReader(void* task) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)task;
SOperatorInfo* pOp = pTaskInfo->pRoot;
qDebug("stream close tsdb reader, reset status uid:%" PRId64 " ts:%" PRId64, pTaskInfo->streamInfo.lastStatus.uid,
pTaskInfo->streamInfo.lastStatus.ts);
qDebug("stream close tsdb reader, reset status uid:%" PRId64 " ts:%" PRId64, pTaskInfo->streamInfo.currentOffset.uid,
pTaskInfo->streamInfo.currentOffset.ts);
// todo refactor, other thread may already use this read to extract data.
pTaskInfo->streamInfo.lastStatus = (STqOffsetVal){0};
pTaskInfo->streamInfo.currentOffset = (STqOffsetVal){0};
while (pOp->numOfDownstream == 1 && pOp->pDownstream[0]) {
SOperatorInfo* pDownstreamOp = pOp->pDownstream[0];
if (pDownstreamOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
......
source/libs/executor/src/projectoperator.c
浏览文件 @ 2b2059ef
......@@ -227,17 +227,8 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
blockDataCleanup(pFinalRes);
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pTaskInfo->streamInfo.submit.msgStr) {
pOperator->status = OP_OPENED;
}
if (pOperator->status == OP_EXEC_DONE) {
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
pOperator->status = OP_OPENED;
qDebug("projection in queue model, set status open and return null");
return NULL;
}
return NULL;
}
......@@ -263,23 +254,14 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// The downstream exec may change the value of the newgroup, so use a local variable instead.
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) {
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE && pFinalRes->info.rows == 0) {
pOperator->status = OP_OPENED;
if (pOperator->status == OP_EXEC_RECV) {
continue;
} else {
return NULL;
}
}
qDebug("set op close, exec %d, status %d rows %" PRId64 , pTaskInfo->execModel, pOperator->status,
pFinalRes->info.rows);
qDebug("set op close, exec %d, status %d rows %" PRId64 , pTaskInfo->execModel, pOperator->status, pFinalRes->info.rows);
setOperatorCompleted(pOperator);
break;
}
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
qDebug("set status recv");
pOperator->status = OP_EXEC_RECV;
}
// if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
// qDebug("set status recv");
// pOperator->status = OP_EXEC_RECV;
// }
// for stream interval
if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
......
source/libs/executor/src/scanoperator.c
浏览文件 @ 2b2059ef
......@@ -707,9 +707,9 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
// todo refactor
/*pTableScanInfo->lastStatus.uid = pBlock->info.id.uid;*/
/*pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;*/
pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
// pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
// pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
// pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
return pBlock;
}
......@@ -899,9 +899,11 @@ static void destroyTableScanOperatorInfo(void* param) {
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* readHandle,
STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {
int32_t code = 0;
STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _error;
}
......@@ -909,8 +911,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
int32_t numOfCols = 0;
int32_t code =
extractColMatchInfo(pScanNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->base.matchInfo);
code = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->base.matchInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
......@@ -1632,14 +1633,11 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
qDebug("start to exec queue scan, %s", id);
if (pTaskInfo->streamInfo.submit.msgStr != NULL) {
if (pInfo->tqReader->msg2.msgStr == NULL) {
SPackedData submit = pTaskInfo->streamInfo.submit;
if (tqReaderSetSubmitReq2(pInfo->tqReader, submit.msgStr, submit.msgLen, submit.ver) < 0) {
qError("submit msg messed up when initing stream submit block %p, %s", submit.msgStr, id);
pInfo->tqReader->msg2 = (SPackedData){0};
pInfo->tqReader->setMsg = 0;
ASSERT(0);
qError("submit msg messed up when initing stream submit block %p", submit.msgStr);
return NULL;
}
}
......@@ -1662,73 +1660,49 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
}
pInfo->tqReader->msg2 = (SPackedData){0};
pInfo->tqReader->setMsg = 0;
pTaskInfo->streamInfo.submit = (SPackedData){0};
return NULL;
}
if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
if (pResult && pResult->info.rows > 0) {
qDebug("queue scan tsdb return %" PRId64 " rows min:%" PRId64 " max:%" PRId64 " wal curVersion:%" PRId64" %s", pResult->info.rows,
pResult->info.window.skey, pResult->info.window.ekey, pInfo->tqReader->pWalReader->curVersion, id);
pTaskInfo->streamInfo.returned = 1;
qDebug("queue scan tsdb return %"PRId64" rows min:%" PRId64 " max:%" PRId64 " wal curVersion:%" PRId64, pResult->info.rows,
pResult->info.window.skey, pResult->info.window.ekey, pInfo->tqReader->pWalReader->curVersion);
tqOffsetResetToData(&pTaskInfo->streamInfo.currentOffset, pResult->info.id.uid, pResult->info.window.ekey);
return pResult;
} else {
// no data has return already, try to extract data in the WAL
if (!pTaskInfo->streamInfo.returned) {
STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
tsdbReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL;
tqOffsetResetToLog(&pTaskInfo->streamInfo.prepareStatus, pTaskInfo->streamInfo.snapshotVer);
qDebug("queue scan tsdb over, switch to wal ver:%" PRId64 " %s", pTaskInfo->streamInfo.snapshotVer + 1, id);
if (tqSeekVer(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1, pTaskInfo->id.str) < 0) {
tqOffsetResetToLog(&pTaskInfo->streamInfo.lastStatus, pTaskInfo->streamInfo.snapshotVer);
return NULL;
}
} else {
return NULL;
}
}
STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
tsdbReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL;
qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1);
if (tqSeekVer(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1, pTaskInfo->id.str) < 0) {
return NULL;
}
tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pTaskInfo->streamInfo.snapshotVer);
}
if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) {
while (1) {
SFetchRet ret = {0};
terrno = 0;
if (tqNextBlock(pInfo->tqReader, &ret) < 0) {
// if the end is reached, terrno is 0
if (terrno != 0) {
qError("failed to get next log block since %s, %s", terrstr(), id);
}
}
tqNextBlock(pInfo->tqReader, &ret);
tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pInfo->tqReader->pWalReader->curVersion - 1); //curVersion move to next, so currentOffset = curVersion - 1
if (ret.fetchType == FETCH_TYPE__DATA) {
qDebug("doQueueScan get data from log %"PRId64" rows, version:%" PRId64, ret.data.info.rows, pTaskInfo->streamInfo.currentOffset.version);
blockDataCleanup(pInfo->pRes);
setBlockIntoRes(pInfo, &ret.data, true);
if (pInfo->pRes->info.rows > 0) {
pOperator->status = OP_EXEC_RECV;
qDebug("queue scan log return %" PRId64 " rows", pInfo->pRes->info.rows);
qDebug("doQueueScan get data from log %"PRId64" rows, return, version:%" PRId64, pInfo->pRes->info.rows, pTaskInfo->streamInfo.currentOffset.version);
return pInfo->pRes;
}
} else if (ret.fetchType == FETCH_TYPE__META) {
qError("unexpected ret.fetchType:%d", ret.fetchType);
continue;
} else if (ret.fetchType == FETCH_TYPE__NONE ||
(ret.fetchType == FETCH_TYPE__SEP && pOperator->status == OP_EXEC_RECV)) {
pTaskInfo->streamInfo.lastStatus = ret.offset;
char formatBuf[80];
tFormatOffset(formatBuf, 80, &ret.offset);
qDebug("queue scan log return null, offset %s", formatBuf);
pOperator->status = OP_OPENED;
}else if(ret.fetchType == FETCH_TYPE__NONE){
qDebug("doQueueScan get none from log, return, version:%" PRId64, pTaskInfo->streamInfo.currentOffset.version);
return NULL;
}
}
} else {
qError("unexpected streamInfo prepare type: %d %s", pTaskInfo->streamInfo.prepareStatus.type, id);
qError("unexpected streamInfo prepare type: %d", pTaskInfo->streamInfo.currentOffset.type);
return NULL;
}
}
......@@ -2083,7 +2057,6 @@ FETCH_NEXT_BLOCK:
int32_t current = pInfo->validBlockIndex++;
SPackedData* pSubmit = taosArrayGet(pInfo->pBlockLists, current);
/*if (tqReaderSetDataMsg(pInfo->tqReader, pSubmit, 0) < 0) {*/
if (tqReaderSetSubmitReq2(pInfo->tqReader, pSubmit->msgStr, pSubmit->msgLen, pSubmit->ver) < 0) {
qError("submit msg messed up when initing stream submit block %p, current %d, total %d", pSubmit, current,
totBlockNum);
......@@ -2175,7 +2148,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
pTaskInfo->streamInfo.metaRsp.metaRsp = NULL;
qDebug("tmqsnap doRawScan called");
if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
bool hasNext = false;
if (pInfo->dataReader) {
code = tsdbNextDataBlock(pInfo->dataReader, &hasNext);
......@@ -2197,28 +2170,23 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
}
qDebug("tmqsnap doRawScan get data uid:%" PRId64 "", pBlock->info.id.uid);
pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_DATA;
pTaskInfo->streamInfo.lastStatus.uid = pBlock->info.id.uid;
pTaskInfo->streamInfo.lastStatus.ts = pBlock->info.window.ekey;
tqOffsetResetToData(&pTaskInfo->streamInfo.currentOffset, pBlock->info.id.uid, pBlock->info.window.ekey);
return pBlock;
}
SMetaTableInfo mtInfo = getUidfromSnapShot(pInfo->sContext);
STqOffsetVal offset = {0};
if (mtInfo.uid == 0) { // read snapshot done, change to get data from wal
qDebug("tmqsnap read snapshot done, change to get data from wal");
pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
pTaskInfo->streamInfo.lastStatus.version = pInfo->sContext->snapVersion;
tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion);
} else {
pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
pTaskInfo->streamInfo.prepareStatus.ts = INT64_MIN;
tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN);
qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);
qStreamPrepareScan(pTaskInfo, &pTaskInfo->streamInfo.prepareStatus, pInfo->sContext->subType);
}
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
tDeleteSSchemaWrapper(mtInfo.schema);
qDebug("tmqsnap stream scan tsdb return null");
return NULL;
} else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) {
} else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {
SSnapContext* sContext = pInfo->sContext;
void* data = NULL;
int32_t dataLen = 0;
......@@ -2230,16 +2198,12 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
return NULL;
}
if (!sContext->queryMetaOrData) { // change to get data next poll request
pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_META;
pTaskInfo->streamInfo.lastStatus.uid = uid;
pTaskInfo->streamInfo.metaRsp.rspOffset.type = TMQ_OFFSET__SNAPSHOT_DATA;
pTaskInfo->streamInfo.metaRsp.rspOffset.uid = 0;
pTaskInfo->streamInfo.metaRsp.rspOffset.ts = INT64_MIN;
if (!sContext->queryMeta) { // change to get data next poll request
STqOffsetVal offset = {0};
tqOffsetResetToData(&offset, 0, INT64_MIN);
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
} else {
pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__SNAPSHOT_META;
pTaskInfo->streamInfo.lastStatus.uid = uid;
pTaskInfo->streamInfo.metaRsp.rspOffset = pTaskInfo->streamInfo.lastStatus;
tqOffsetResetToMeta(&pTaskInfo->streamInfo.currentOffset, uid);
pTaskInfo->streamInfo.metaRsp.resMsgType = type;
pTaskInfo->streamInfo.metaRsp.metaRspLen = dataLen;
pTaskInfo->streamInfo.metaRsp.metaRsp = data;
......@@ -2453,7 +2417,6 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
if (pHandle->initTableReader) {
pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER;
pTSInfo->base.dataReader = NULL;
pTaskInfo->streamInfo.lastStatus.uid = -1;
}
if (pHandle->initTqReader) {
......
source/libs/parser/src/parInsertUtil.c
浏览文件 @ 2b2059ef
......@@ -470,11 +470,11 @@ int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks) {
if (TSDB_CODE_SUCCESS == code) {
SVgroupDataCxt* pVgCxt = NULL;
int32_t vgId = pTableCxt->pMeta->vgId;
void** p = taosHashGet(pVgroupHash, &vgId, sizeof(vgId));
if (NULL == p) {
void** pp = taosHashGet(pVgroupHash, &vgId, sizeof(vgId));
if (NULL == pp) {
code = createVgroupDataCxt(pTableCxt, pVgroupHash, pVgroupList, &pVgCxt);
} else {
pVgCxt = *(SVgroupDataCxt**)p;
pVgCxt = *(SVgroupDataCxt**)pp;
}
if (TSDB_CODE_SUCCESS == code) {
code = fillVgroupDataCxt(pTableCxt, pVgCxt);
......
source/libs/wal/src/walRead.c
浏览文件 @ 2b2059ef
......@@ -33,7 +33,6 @@ SWalReader *walOpenReader(SWal *pWal, SWalFilterCond *cond) {
pReader->pLogFile = NULL;
pReader->curVersion = -1;
pReader->curFileFirstVer = -1;
pReader->curInvalid = 1;
pReader->capacity = 0;
if (cond) {
pReader->cond = *cond;
......@@ -81,7 +80,6 @@ int32_t walNextValidMsg(SWalReader *pReader) {
wDebug("vgId:%d, wal start to fetch, index:%" PRId64 ", last index:%" PRId64 " commit index:%" PRId64
", applied index:%" PRId64 ", end index:%" PRId64,
pReader->pWal->cfg.vgId, fetchVer, lastVer, committedVer, appliedVer, endVer);
pReader->curStopped = 0;
while (fetchVer <= endVer) {
if (walFetchHeadNew(pReader, fetchVer) < 0) {
return -1;
......@@ -99,7 +97,6 @@ int32_t walNextValidMsg(SWalReader *pReader) {
fetchVer = pReader->curVersion;
}
}
pReader->curStopped = 1;
return -1;
}
......@@ -196,17 +193,16 @@ int32_t walReadSeekVerImpl(SWalReader *pReader, int64_t ver) {
return -1;
}
wDebug("vgId:%d, wal version reset from %" PRId64 "(invalid:%d) to %" PRId64, pReader->pWal->cfg.vgId,
pReader->curVersion, pReader->curInvalid, ver);
wDebug("vgId:%d, wal version reset from %" PRId64 " to %" PRId64, pReader->pWal->cfg.vgId,
pReader->curVersion, ver);
pReader->curVersion = ver;
pReader->curInvalid = 0;
return 0;
}
int32_t walReadSeekVer(SWalReader *pReader, int64_t ver) {
SWal *pWal = pReader->pWal;
if (!pReader->curInvalid && ver == pReader->curVersion) {
if (ver == pReader->curVersion) {
wDebug("vgId:%d, wal index:%" PRId64 " match, no need to reset", pReader->pWal->cfg.vgId, ver);
return 0;
}
......@@ -238,7 +234,7 @@ static int32_t walFetchHeadNew(SWalReader *pRead, int64_t fetchVer) {
wDebug("vgId:%d, wal starts to fetch head, index:%" PRId64, pRead->pWal->cfg.vgId, fetchVer);
if (pRead->curInvalid || pRead->curVersion != fetchVer) {
if (pRead->curVersion != fetchVer) {
if (walReadSeekVer(pRead, fetchVer) < 0) {
// pRead->curVersion = fetchVer;
// pRead->curInvalid = 1;
......@@ -344,7 +340,7 @@ int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead) {
return -1;
}
if (pRead->curInvalid || pRead->curVersion != ver) {
if (pRead->curVersion != ver) {
code = walReadSeekVer(pRead, ver);
if (code < 0) {
// pRead->curVersion = ver;
......@@ -479,7 +475,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
taosThreadMutexLock(&pReader->mutex);
if (pReader->curInvalid || pReader->curVersion != ver) {
if (pReader->curVersion != ver) {
if (walReadSeekVer(pReader, ver) < 0) {
wError("vgId:%d, unexpected wal log, index:%" PRId64 ", since %s", pReader->pWal->cfg.vgId, ver, terrstr());
taosThreadMutexUnlock(&pReader->mutex);
......@@ -575,7 +571,6 @@ void walReadReset(SWalReader *pReader) {
taosThreadMutexLock(&pReader->mutex);
taosCloseFile(&pReader->pIdxFile);
taosCloseFile(&pReader->pLogFile);
pReader->curInvalid = 1;
pReader->curFileFirstVer = -1;
pReader->curVersion = -1;
taosThreadMutexUnlock(&pReader->mutex);
......
source/util/src/terror.c
浏览文件 @ 2b2059ef
......@@ -628,6 +628,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_INVALID_FILE, "Index file is inval
TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_INVALID_MSG, "Invalid message")
TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_MISMATCH, "Consumer mismatch")
TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_CLOSED, "Consumer closed")
TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_ERROR, "Consumer error, to see log")
// stream
TAOS_DEFINE_ERROR(TSDB_CODE_STREAM_TASK_NOT_EXIST, "Stream task not exist")
......
tests/parallel_test/cases.task
浏览文件 @ 2b2059ef
......@@ -94,7 +94,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropNtb-snapshot0.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropNtb-snapshot1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUdf.py
#,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUdf.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUdf-multCtb-snapshot0.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUdf-multCtb-snapshot1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/stbTagFilter-1ctb.py
......
tests/script/tsim/tmq/basic1.sim
浏览文件 @ 2b2059ef
......@@ -111,8 +111,8 @@ endi
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
......@@ -134,7 +134,7 @@ if $data[0][2] != $expectmsgcnt then
print expect $expectmsgcnt , actual $data02
return -1
endi
if $data[0][3] != $expectmsgcnt then
if $data[0][3] != $totalMsgOfStb then
return -1
endi
$loop_cnt = $loop_cnt + 1
......@@ -183,8 +183,8 @@ endi
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -202,7 +202,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfCtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfCtb then
......@@ -254,8 +254,8 @@ endi
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -273,7 +273,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfNtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfNtb then
......
tests/script/tsim/tmq/basic1Of2Cons.sim
浏览文件 @ 2b2059ef
......@@ -111,11 +111,11 @@ endi
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
......@@ -142,14 +142,14 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfStb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfStb
if $data[0][2] == $totalMsgOfStb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_0
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfStb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_0
endi
endi
......@@ -214,10 +214,10 @@ endi
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -244,14 +244,14 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfCtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfCtb
if $data[0][2] == $totalMsgOfCtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_2
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfCtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_2
endi
endi
......@@ -316,10 +316,10 @@ endi
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -346,14 +346,14 @@ if $data[1][1] == 0 then
endi
endi
# either $data[0][2] == $totalMsgOfNtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfNtb
if $data[0][2] == $totalMsgOfNtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_4
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfNtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_4
endi
endi
......
tests/script/tsim/tmq/basic2.sim
浏览文件 @ 2b2059ef
......@@ -83,8 +83,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$totalMsgOfStb = $totalMsgOfStb * $topicNum
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
......@@ -105,7 +105,7 @@ endi
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $expectmsgcnt then
if $data[0][3] != $totalMsgOfStb then
return -1
endi
......@@ -140,8 +140,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -159,7 +159,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfCtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfCtb then
......@@ -197,8 +197,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -216,7 +216,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfNtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfNtb then
......
tests/script/tsim/tmq/basic2Of2Cons.sim
浏览文件 @ 2b2059ef
......@@ -82,10 +82,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$totalMsgOfStb = $totalMsgOfStb * $topicNum
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
......@@ -112,14 +112,14 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfStb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfStb
if $data[0][2] == $totalMsgOfStb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_0
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfStb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_0
endi
endi
......@@ -169,10 +169,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -199,14 +199,14 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfCtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfCtb
if $data[0][2] == $totalMsgOfCtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_2
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfCtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_2
endi
endi
......@@ -256,10 +256,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -286,14 +286,14 @@ if $data[1][1] == 0 then
endi
endi
# either $data[0][2] == $totalMsgOfNtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfNtb
if $data[0][2] == $totalMsgOfNtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_4
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfNtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_4
endi
endi
......
tests/script/tsim/tmq/basic2Of2ConsOverlap.sim
浏览文件 @ 2b2059ef
......@@ -80,8 +80,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfOneTopic = $ctbNum * $rowsPerCtb
$totalMsgOfStb = $totalMsgOfOneTopic * $topicNum
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 3
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
$topicList = ' . topic_stb_all
......@@ -89,7 +89,7 @@ $topicList = $topicList . ,
$topicList = $topicList . topic_stb_function
$topicList = $topicList . '
$consumerId = 1
sql insert into consumeinfo values (now +1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now +1s , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
......@@ -118,40 +118,40 @@ endi
# $data[0][2]/$data[1][2] should be between $totalMsgOfOneTopic and $totalMsgOfStb.
if $data[0][2] < $totalMsgOfOneTopic then
return -1
endi
if $data[0][2] > $totalMsgOfStb then
return -1
endi
if $data[1][2] < $totalMsgOfOneTopic then
#if $data[0][2] < $totalMsgOfOneTopic then
# return -1
#endi
if $data[0][2] > $expectmsgcnt then
return -1
endi
if $data[1][2] > $totalMsgOfStb then
#if $data[1][2] < $totalMsgOfOneTopic then
# return -1
#endi
if $data[1][2] > $expectmsgcnt then
return -1
endi
$totalMsgCons = $totalMsgOfOneTopic + $totalMsgOfStb
#$totalMsgCons = $totalMsgOfOneTopic + $totalMsgOfStb
$sumOfMsgCnt = $data[0][2] + $data[1][2]
if $sumOfMsgCnt != $totalMsgCons then
print total: $totalMsgCons
if $sumOfMsgCnt != $expectmsgcnt then
print total: $expectmsgcnt
print sum: $sumOfMsgCnt
return -1
endi
# $data[0][3]/$data[1][3] should be between $totalMsgOfOneTopic and $totalMsgOfStb.
if $data[0][3] < $totalMsgOfOneTopic then
return -1
endi
if $data[0][3] > $totalMsgOfStb then
return -1
endi
if $data[1][3] < $totalMsgOfOneTopic then
return -1
endi
if $data[1][3] > $totalMsgOfStb then
return -1
endi
#if $data[0][3] < $totalMsgOfStb then
# return -1
#endi
#if $data[0][3] > $totalMsgOfStb then
# return -1
#endi
#if $data[1][3] < $totalMsgOfStb then
# return -1
#endi
#if $data[1][3] > $totalMsgOfStb then
# return -1
#endi
$totalMsgCons = $totalMsgOfOneTopic + $totalMsgOfStb
$sumOfRows = $data[0][3] + $data[1][3]
......@@ -189,15 +189,15 @@ $consumerId = 0
$totalMsgOfOneTopic = $rowsPerCtb
$totalMsgOfCtb = $totalMsgOfOneTopic * $topicNum
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
$topicList = ' . topic_ctb_function
$topicList = $topicList . ,
$topicList = $topicList . topic_ctb_all
$topicList = $topicList . '
$consumerId = 1
sql insert into consumeinfo values (now +1s, $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now +1s, $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
......@@ -226,24 +226,24 @@ endi
# either $data[0][2] $totalMsgOfOneTopic and $data[1][2] == $totalMsgOfCtb
# or $data[0][2] $totalMsgOfCtb and $data[1][2] == $totalMsgOfOneTopic
if $data[0][2] == $totalMsgOfOneTopic then
if $data[1][2] == $totalMsgOfCtb then
if $data[0][2] == $topicNum then
if $data[1][2] == 1 then
goto check_ok_0
endi
elif $data[1][2] == $totalMsgOfOneTopic then
if $data[0][2] == $totalMsgOfCtb then
elif $data[0][2] == 1 then
if $data[1][2] == $topicNum then
goto check_ok_0
endi
endi
return -1
check_ok_0:
if $data[0][3] == $totalMsgOfOneTopic then
if $data[1][3] == $totalMsgOfCtb then
if $data[0][3] == $totalMsgOfCtb then
if $data[1][3] == $totalMsgOfOneTopic then
goto check_ok_1
endi
elif $data[1][3] == $totalMsgOfOneTopic then
if $data[0][3] == $totalMsgOfCtb then
elif $data[0][3] == $totalMsgOfOneTopic then
if $data[1][3] == $totalMsgOfCtb then
goto check_ok_1
endi
endi
......@@ -280,8 +280,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfOneTopic = $rowsPerCtb
$totalMsgOfNtb = $totalMsgOfOneTopic * $topicNum
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
$topicList = ' . topic_ntb_function
......@@ -289,7 +289,7 @@ $topicList = $topicList . ,
$topicList = $topicList . topic_ntb_all
$topicList = $topicList . '
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -318,12 +318,12 @@ endi
# either $data[0][2] $totalMsgOfOneTopic and $data[1][2] == $totalMsgOfNtb
# or $data[0][2] $totalMsgOfNtb and $data[1][2] == $totalMsgOfOneTopic
if $data[0][2] == $totalMsgOfOneTopic then
if $data[1][2] == $totalMsgOfNtb then
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 1 then
goto check_ok_2
endi
elif $data[1][2] == $totalMsgOfOneTopic then
if $data[0][2] == $totalMsgOfNtb then
elif $data[0][2] == 1 then
if $data[1][2] == $expectmsgcnt then
goto check_ok_2
endi
endi
......@@ -334,8 +334,8 @@ if $data[0][3] == $totalMsgOfOneTopic then
if $data[1][3] == $totalMsgOfNtb then
goto check_ok_3
endi
elif $data[1][3] == $totalMsgOfOneTopic then
if $data[0][3] == $totalMsgOfNtb then
elif $data[0][3] == $totalMsgOfNtb then
if $data[1][3] == $totalMsgOfOneTopic then
goto check_ok_3
endi
endi
......
tests/script/tsim/tmq/basic3.sim
浏览文件 @ 2b2059ef
......@@ -111,8 +111,8 @@ endi
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
......@@ -130,10 +130,10 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $expectmsgcnt then
#if $data[0][2] != $expectmsgcnt then
# return -1
#endi
if $data[0][3] != $totalMsgOfStb then
return -1
endi
$loop_cnt = $loop_cnt + 1
......@@ -182,8 +182,8 @@ endi
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -201,7 +201,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfCtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfCtb then
......@@ -253,8 +253,8 @@ endi
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -272,7 +272,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfNtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfNtb then
......
tests/script/tsim/tmq/basic3Of2Cons.sim
浏览文件 @ 2b2059ef
......@@ -110,10 +110,10 @@ endi
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 3
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start
......@@ -163,19 +163,19 @@ endi
if $data[0][3] <= 0 then
return -1
endi
if $data[0][3] >= $expectmsgcnt then
if $data[0][3] >= $totalMsgOfStb then
return -1
endi
if $data[1][3] <= 0 then
return -1
endi
if $data[1][3] >= $expectmsgcnt then
if $data[1][3] >= $totalMsgOfStb then
return -1
endi
$sumOfMsgRows = $data[0][3] + $data[1][3]
if $sumOfMsgRows != $expectmsgcnt then
if $sumOfMsgRows != $totalMsgOfStb then
return -1
endi
......@@ -225,10 +225,10 @@ endi
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -255,13 +255,13 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfCtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfCtb
if $data[0][2] == $totalMsgOfCtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_0
endi
elif $data[1][2] == $totalMsgOfCtb then
elif $data[1][2] == $expectmsgcnt then
if $data[0][2] == 0 then
goto check_ok_0
endi
......@@ -327,10 +327,10 @@ endi
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -357,13 +357,13 @@ if $data[1][1] == 0 then
endi
endi
# either $data[0][2] == $totalMsgOfNtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfNtb
if $data[0][2] == $totalMsgOfNtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_2
endi
elif $data[1][2] == $totalMsgOfNtb then
elif $data[1][2] == $expectmsgcnt then
if $data[0][2] == 0 then
goto check_ok_2
endi
......
tests/script/tsim/tmq/basic4.sim
浏览文件 @ 2b2059ef
......@@ -80,8 +80,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$totalMsgOfStb = $totalMsgOfStb * $topicNum
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 9
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
......@@ -102,7 +102,7 @@ endi
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $expectmsgcnt then
if $data[0][3] != $totalMsgOfStb then
return -1
endi
......@@ -137,8 +137,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -156,7 +156,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfCtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfCtb then
......@@ -194,8 +194,8 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -213,7 +213,7 @@ endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $totalMsgOfNtb then
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $totalMsgOfNtb then
......
tests/script/tsim/tmq/basic4Of2Cons.sim
浏览文件 @ 2b2059ef
......@@ -79,10 +79,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$totalMsgOfStb = $totalMsgOfStb * $topicNum
$expectmsgcnt = $totalMsgOfStb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = 9
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfStb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $dbName -s start
......@@ -131,19 +131,19 @@ endi
if $data[0][3] <= 0 then
return -1
endi
if $data[0][3] >= $expectmsgcnt then
if $data[0][3] >= $totalMsgOfStb then
return -1
endi
if $data[1][3] <= 0 then
return -1
endi
if $data[1][3] >= $expectmsgcnt then
if $data[1][3] >= $totalMsgOfStb then
return -1
endi
$sumOfConsRow = $data[0][3] + $data[1][3]
if $sumOfConsRow != $expectmsgcnt then
if $sumOfConsRow != $totalMsgOfStb then
return -1
endi
......@@ -178,10 +178,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfCtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfCtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -208,14 +208,14 @@ if $data[0][1] == 1 then
endi
endi
# either $data[0][2] == $totalMsgOfCtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfCtb
if $data[0][2] == $totalMsgOfCtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_0
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfCtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_0
endi
endi
......@@ -266,10 +266,10 @@ $topicList = $topicList . '
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb * $topicNum
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
$expectmsgcnt = $topicNum
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
$consumerId = 1
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
sql insert into consumeinfo values (now+1s , $consumerId , $topicList , $keyList , $totalMsgOfNtb , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start
......@@ -296,14 +296,14 @@ if $data[1][1] == 0 then
endi
endi
# either $data[0][2] == $totalMsgOfNtb and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $totalMsgOfNtb
if $data[0][2] == $totalMsgOfNtb then
# either $data[0][2] == $expectmsgcnt and $data[1][2] == 0
# or $data[0][2] == 0 and $data[1][2] == $expectmsgcnt
if $data[0][2] == $expectmsgcnt then
if $data[1][2] == 0 then
goto check_ok_2
endi
elif $data[0][2] == 0 then
if $data[1][2] == $totalMsgOfNtb then
if $data[1][2] == $expectmsgcnt then
goto check_ok_2
endi
endi
......
tests/system-test/7-tmq/subscribeStb.py
浏览文件 @ 2b2059ef
......@@ -226,7 +226,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -307,7 +307,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......
tests/system-test/7-tmq/subscribeStb0.py
浏览文件 @ 2b2059ef
......@@ -228,7 +228,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -303,7 +303,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -315,7 +315,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt/4:
if totalConsumeRows < expectrowcnt/4:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
tdLog.exit("tmq consume rows error!")
......@@ -333,7 +333,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt:
if totalConsumeRows < expectrowcnt:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt))
tdLog.exit("tmq consume rows error!")
......@@ -386,7 +386,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -398,7 +398,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt/4:
if totalConsumeRows < expectrowcnt/4:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
tdLog.exit("tmq consume rows error!")
......@@ -416,7 +416,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != (expectrowcnt * (1 + 1/4)):
if totalConsumeRows < (expectrowcnt * (1 + 1/4)):
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt))
tdLog.exit("tmq consume rows error!")
......
tests/system-test/7-tmq/subscribeStb1.py
浏览文件 @ 2b2059ef
......@@ -233,7 +233,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -320,7 +320,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......
tests/system-test/7-tmq/subscribeStb2.py
浏览文件 @ 2b2059ef
......@@ -233,7 +233,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume 0 processor")
pollDelay = 100
pollDelay = 10
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -251,6 +251,7 @@ class TDTestCase:
tdLog.info("start consume 1 processor")
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
tdLog.sleep(2)
tdLog.info("start one new thread to insert data")
parameterDict['actionType'] = actionType.INSERT_DATA
......@@ -271,6 +272,7 @@ class TDTestCase:
tdLog.info("start consume 2 processor")
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
tdLog.sleep(2)
tdLog.info("start one new thread to insert data")
parameterDict['actionType'] = actionType.INSERT_DATA
......@@ -338,7 +340,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume 0 processor")
pollDelay = 100
pollDelay = 20
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -380,6 +382,7 @@ class TDTestCase:
tdLog.info("start consume 2 processor")
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
tdLog.sleep(2)
tdLog.info("start one new thread to insert data")
parameterDict['actionType'] = actionType.INSERT_DATA
......@@ -394,7 +397,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt*2:
if totalConsumeRows < expectrowcnt*2:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt*2))
tdLog.exit("tmq consume rows error!")
......
tests/system-test/7-tmq/subscribeStb3.py
浏览文件 @ 2b2059ef
......@@ -215,7 +215,8 @@ class TDTestCase:
parameterDict["stbName"],\
parameterDict["ctbNum"],\
parameterDict["rowsPerTbl"],\
parameterDict["batchNum"])
parameterDict["batchNum"],\
parameterDict["startTs"])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
......@@ -233,7 +234,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume 0 processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -269,7 +270,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt-10000:
if totalConsumeRows < expectrowcnt-10000:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt-10000))
tdLog.exit("tmq consume rows error!")
......@@ -328,7 +329,8 @@ class TDTestCase:
parameterDict["stbName"],\
parameterDict["ctbNum"],\
parameterDict["rowsPerTbl"],\
parameterDict["batchNum"])
parameterDict["batchNum"],\
parameterDict["startTs"])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
......@@ -346,7 +348,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -415,7 +417,8 @@ class TDTestCase:
parameterDict["stbName"],\
parameterDict["ctbNum"],\
parameterDict["rowsPerTbl"],\
parameterDict["batchNum"])
parameterDict["batchNum"],\
parameterDict["startTs"])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
......@@ -433,7 +436,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -445,7 +448,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt/4:
if totalConsumeRows < expectrowcnt/4:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
tdLog.exit("tmq consume rows error!")
......@@ -467,7 +470,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt/4:
if totalConsumeRows < expectrowcnt/4:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
tdLog.exit("tmq consume rows error!")
......@@ -502,7 +505,8 @@ class TDTestCase:
parameterDict["stbName"],\
parameterDict["ctbNum"],\
parameterDict["rowsPerTbl"],\
parameterDict["batchNum"])
parameterDict["batchNum"],\
parameterDict["startTs"])
tdLog.info("create topics from stb1")
topicFromStb1 = 'topic_stb1'
......@@ -520,7 +524,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt/4,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -532,7 +536,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt/4:
if totalConsumeRows < expectrowcnt/4:
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4))
tdLog.exit("tmq consume rows error!")
......@@ -555,7 +559,7 @@ class TDTestCase:
for i in range(expectRows):
totalConsumeRows += resultList[i]
if totalConsumeRows != expectrowcnt*(1/2+1/4):
if totalConsumeRows < expectrowcnt*(1/2+1/4):
tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt*(1/2+1/4)))
tdLog.exit("tmq consume rows error!")
......
tests/system-test/7-tmq/subscribeStb4.py
浏览文件 @ 2b2059ef
......@@ -231,7 +231,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......@@ -305,7 +305,7 @@ class TDTestCase:
self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
pollDelay = 100
pollDelay = 5
showMsg = 1
showRow = 1
self.startTmqSimProcess(buildPath,cfgPath,pollDelay,parameterDict["dbName"],showMsg, showRow)
......
tests/system-test/7-tmq/tmqDelete-1ctb.py
浏览文件 @ 2b2059ef
......@@ -53,7 +53,7 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1,wal_retention_size=-1, wal_retention_period=-1)
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
......@@ -237,10 +237,10 @@ class TDTestCase:
if self.snapshot == 0:
consumerId = 2
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 + 1/4 + 3/4))
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"])
elif self.snapshot == 1:
consumerId = 3
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 - 1/4 + 1/4 + 3/4))
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 + 1/4))
topicList = topicFromStb1
ifcheckdata = 1
......@@ -270,7 +270,7 @@ class TDTestCase:
if totalConsumeRows != expectrowcnt:
tdLog.exit("tmq consume rows error with snapshot = 0!")
elif self.snapshot == 1:
if totalConsumeRows != totalRowsFromQuery:
if totalConsumeRows != expectrowcnt:
tdLog.exit("tmq consume rows error with snapshot = 1!")
# tmqCom.checkFileContent(consumerId, queryString)
......@@ -323,7 +323,7 @@ class TDTestCase:
if self.snapshot == 0:
consumerId = 4
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1/4 + 3/4))
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"])
elif self.snapshot == 1:
consumerId = 5
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 - 1/4 + 1/4 + 3/4))
......@@ -369,11 +369,7 @@ class TDTestCase:
tdLog.info("act consume rows: %d, act query rows: %d, expect consume rows: %d, "%(totalConsumeRows, totalRowsFromQuery, expectrowcnt))
if self.snapshot == 0:
# If data writing is completed before consumer get snapshot, will consume 7500 from wal;
# If data writing has not started before consumer get snapshot, will consume 10000 from wal;
minRows = int(expectrowcnt * (1 - 1/4)) # 7500
tdLog.info("consume rows should be between %d and %d, "%(minRows, expectrowcnt))
if not ((totalConsumeRows >= minRows) and (totalConsumeRows <= expectrowcnt)):
if (totalConsumeRows != expectrowcnt):
tdLog.exit("tmq consume rows error with snapshot = 0!")
elif self.snapshot == 1:
tdLog.info("consume rows should be between %d and %d, "%(totalRowsFromQuery, expectrowcnt))
......@@ -494,7 +490,7 @@ class TDTestCase:
tdLog.printNoPrefix("======== test case 4 end ...... ")
def run(self):
# tdSql.prepare()
tdSql.prepare()
tdLog.printNoPrefix("=============================================")
tdLog.printNoPrefix("======== snapshot is 0: only consume from wal")
self.snapshot = 0
......@@ -520,11 +516,11 @@ class TDTestCase:
self.prepareTestEnv()
self.tmqCase3()
tdLog.printNoPrefix("=============================================")
tdLog.printNoPrefix("======== snapshot is 0: only consume from wal")
self.snapshot = 0
self.prepareTestEnv()
self.tmqCase4()
# tdLog.printNoPrefix("=============================================")
# tdLog.printNoPrefix("======== snapshot is 0: only consume from wal")
# self.snapshot = 0
# self.prepareTestEnv()
# self.tmqCase4()
tdLog.printNoPrefix("====================================================================")
tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal")
self.snapshot = 1
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或