diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 7e106eefdeca155bc8c8d4ce139cdd8b50ce1460..7af218b78e5daf92a1f37b9ad56ab07ae29433c5 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -205,7 +205,8 @@ int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead); int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead); int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead); -SWalRef *walRefFirstVer(SWal *, SWalRef *); +void walRefFirstVer(SWal *, SWalRef *); +void walRefLastVer(SWal *, SWalRef *); SWalRef *walRefCommittedVer(SWal *); SWalRef *walOpenRef(SWal *); diff --git a/source/dnode/vnode/src/tq/tqUtil.c b/source/dnode/vnode/src/tq/tqUtil.c index 33d2a238c5507646470569028b7d850e3f8474e0..5eaf7b240b37520ea06533fef5da2e5bd0ba18c6 100644 --- a/source/dnode/vnode/src/tq/tqUtil.c +++ b/source/dnode/vnode/src/tq/tqUtil.c @@ -115,18 +115,12 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand tqOffsetResetToData(pOffsetVal, 0, 0); } } else { - pHandle->pRef = walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef); - if (pHandle->pRef == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return -1; - } - - // offset set to previous version when init + walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef); tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1); } } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) { - // offset set to previous version when init - tqOffsetResetToLog(pOffsetVal, walGetLastVer(pTq->pVnode->pWal)); + walRefLastVer(pTq->pVnode->pWal, pHandle->pRef); + tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer); } else if (reqOffset.type == TMQ_OFFSET__RESET_NONE) { tqError("tmq poll: subkey:%s, no offset committed for consumer:0x%" PRIx64 " in vg %d, subkey %s, reset none failed", pHandle->subKey, consumerId, vgId, pRequest->subKey); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index fb0b9957f850db54bb09c22d17dbd2937f32e2ea..a73deffa528c71999e036370b3a3b021b861f024 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -1080,9 +1080,6 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT const char* id = GET_TASKID(pTaskInfo); // if pOffset equal to current offset, means continue consume - char buf[80] = {0}; - tFormatOffset(buf, 80, &pTaskInfo->streamInfo.currentOffset); - qDebug("currentOffset:%s", buf); if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) { return 0; } diff --git a/source/libs/wal/src/walRef.c b/source/libs/wal/src/walRef.c index 6aba66192684e262baddf4215d043a875c15f6fc..eb36389f1d22c21c828a1c79bdad5ab1deeedbfc 100644 --- a/source/libs/wal/src/walRef.c +++ b/source/libs/wal/src/walRef.c @@ -63,21 +63,22 @@ int32_t walSetRefVer(SWalRef *pRef, int64_t ver) { return 0; } -SWalRef *walRefFirstVer(SWal *pWal, SWalRef *pRef) { - if (pRef == NULL) { - pRef = walOpenRef(pWal); - if (pRef == NULL) { - return NULL; - } - } +void walRefFirstVer(SWal *pWal, SWalRef *pRef) { taosThreadMutexLock(&pWal->mutex); int64_t ver = walGetFirstVer(pWal); pRef->refVer = ver; taosThreadMutexUnlock(&pWal->mutex); wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver); +} - return pRef; +void walRefLastVer(SWal *pWal, SWalRef *pRef) { + taosThreadMutexLock(&pWal->mutex); + int64_t ver = walGetLastVer(pWal); + pRef->refVer = ver; + + taosThreadMutexUnlock(&pWal->mutex); + wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, ver); } SWalRef *walRefCommittedVer(SWal *pWal) {