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

fix(tmq): reset offset

This commit is contained in:
Liu Jicong 2022-07-12 16:39:05 +08:00
parent 7cf99d5359
commit 0fccdace56
3 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -319,9 +319,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
// 3.query
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
if (fetchOffsetNew.type == TMQ_OFFSET__LOG) {
fetchOffsetNew.version++;
}
/*if (fetchOffsetNew.type == TMQ_OFFSET__LOG) {*/
/*fetchOffsetNew.version++;*/
/*}*/
if (tqScan(pTq, pHandle, &dataRsp, &fetchOffsetNew) < 0) {
ASSERT(0);
code = -1;

1
source/dnode/vnode/src/tq/tqExec.c
View File

@ -66,7 +66,6 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVa
if (qStreamPrepareScan(task, pOffset) < 0) {
ASSERT(pOffset->type == TMQ_OFFSET__LOG);
pRsp->rspOffset = *pOffset;
pRsp->rspOffset.version--;
return 0;
}

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

@ -288,14 +288,22 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
if (!tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus)) {
while (1) {
uint8_t type = pOperator->operatorType;
pOperator->status = OP_OPENED;
/*pOperator->status = OP_OPENED;*/
if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
SStreamScanInfo* pInfo = pOperator->info;
if (pOffset->type == TMQ_OFFSET__LOG) {
if (tqSeekVer(pInfo->tqReader, pOffset->version) < 0) {
#if 0
if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus) &&
pInfo->tqReader->pWalReader->curVersion != pOffset->version) {
qError("prepare scan ver %ld actual ver %ld, last %ld", pOffset->version,
pInfo->tqReader->pWalReader->curVersion, pTaskInfo->streamInfo.lastStatus.version);
ASSERT(0);
}
#endif
if (tqSeekVer(pInfo->tqReader, pOffset->version + 1) < 0) {
return -1;
}
ASSERT(pInfo->tqReader->pWalReader->curVersion == pOffset->version);
ASSERT(pInfo->tqReader->pWalReader->curVersion == pOffset->version + 1);
} else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
/*pInfo->blockType = STREAM_INPUT__TABLE_SCAN;*/
int64_t uid = pOffset->uid;