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

add checkpoint

This commit is contained in:
yihaoDeng 2023-06-28 12:22:27 +00:00
parent 0bf6675f39
commit 3d532a54a8
2 changed files with 3 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/tq/tqStreamStateSnap.c
View File

@ -45,7 +45,7 @@ int32_t streamStateSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamS
pReader->ever = ever; pReader->ever = ever;
SStreamSnapReader* pSnapReader = NULL; SStreamSnapReader* pSnapReader = NULL;
sprintf(tdir, "%s%s%s", pTq->path, TD_DIRSEP, VNODE_TQ_STREAM); sprintf(tdir, "%s%s%s%s%s", pTq->path, TD_DIRSEP, VNODE_TQ_STREAM, TD_DIRSEP, "checkpoints");
streamSnapReaderOpen(pTq, sver, ever, tdir, &pSnapReader); streamSnapReaderOpen(pTq, sver, ever, tdir, &pSnapReader);
pReader->pReaderImpl = pSnapReader; pReader->pReaderImpl = pSnapReader;

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

@ -117,10 +117,10 @@ NextTbl:
tdbTbcClose(pReader->pCur); tdbTbcClose(pReader->pCur);
pReader->pos += 1; pReader->pos += 1;
code = tdbTbcOpen(taosArrayGetP(pReader->tdbTbList, pReader->pos), &pReader->pCur, NULL); pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);
code = tdbTbcOpen(pPair->tbl, &pReader->pCur, NULL);
tdbTbcMoveToFirst(pReader->pCur); tdbTbcMoveToFirst(pReader->pCur);
pPair = taosArrayGet(pReader->tdbTbList, pReader->pos);
goto NextTbl; goto NextTbl;
} }
} }
@ -141,8 +141,6 @@ NextTbl:
tqInfo("vgId:%d, vnode stream-task snapshot read data vLen:%d", TD_VID(pReader->pTq->pVnode), vLen); tqInfo("vgId:%d, vnode stream-task snapshot read data vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
return code;
_exit:
return code; return code;
_err: _err:
tqError("vgId:%d, vnode stream-task snapshot read data failed since %s", TD_VID(pReader->pTq->pVnode), tqError("vgId:%d, vnode stream-task snapshot read data failed since %s", TD_VID(pReader->pTq->pVnode),