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

fix(tmq): set last scan status

This commit is contained in:
Liu Jicong 2022-06-30 16:40:17 +08:00
parent 8dece648e3
commit ca1c961f9c
2 changed files with 35 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -85,12 +85,15 @@ int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, S
tqAddBlockDataToRsp(pDataBlock, pRsp); tqAddBlockDataToRsp(pDataBlock, pRsp);
if (pRsp->withTbName) { if (pRsp->withTbName) {
pRsp->withTbName = 0;
#if 1
int64_t uid; int64_t uid;
int64_t ts; int64_t ts;
if (qGetStreamScanStatus(task, &uid, &ts) < 0) { if (qGetStreamScanStatus(task, &uid, &ts) < 0) {
ASSERT(0); ASSERT(0);
} }
tqAddTbNameToRsp(pTq, uid, pRsp, workerId); tqAddTbNameToRsp(pTq, uid, pRsp, workerId);
#endif
} }
pRsp->blockNum++; pRsp->blockNum++;

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

@ -392,6 +392,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
binfo.capacity = binfo.rows; binfo.capacity = binfo.rows;
blockDataEnsureCapacity(pBlock, binfo.rows); blockDataEnsureCapacity(pBlock, binfo.rows);
pBlock->info = binfo; pBlock->info = binfo;
ASSERT(binfo.uid != 0);
uint32_t status = 0; uint32_t status = 0;
int32_t code = loadDataBlock(pOperator, pTableScanInfo, pBlock, &status); int32_t code = loadDataBlock(pOperator, pTableScanInfo, pBlock, &status);
@ -419,6 +420,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
pTableScanInfo->lastStatus.uid = pBlock->info.uid; pTableScanInfo->lastStatus.uid = pBlock->info.uid;
pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey; pTableScanInfo->lastStatus.ts = pBlock->info.window.ekey;
ASSERT(pBlock->info.uid != 0);
return pBlock; return pBlock;
} }
return NULL; return NULL;
@ -438,6 +440,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) { while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
SSDataBlock* p = doTableScanImpl(pOperator); SSDataBlock* p = doTableScanImpl(pOperator);
if (p != NULL) { if (p != NULL) {
ASSERT(p->info.uid != 0);
return p; return p;
} }
pTableScanInfo->curTWinIdx += 1; pTableScanInfo->curTWinIdx += 1;
@ -517,6 +520,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) { if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {
// check status // check status
if (pInfo->lastStatus.uid == pInfo->expStatus.uid && pInfo->lastStatus.ts == pInfo->expStatus.ts) { if (pInfo->lastStatus.uid == pInfo->expStatus.uid && pInfo->lastStatus.ts == pInfo->expStatus.ts) {
while (1) {
SSDataBlock* result = doTableScanGroup(pOperator); SSDataBlock* result = doTableScanGroup(pOperator);
if (result) { if (result) {
return result; return result;
@ -532,9 +536,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
tsdbResetReadHandle(pInfo->dataReader, &pInfo->cond, 0); tsdbResetReadHandle(pInfo->dataReader, &pInfo->cond, 0);
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
pInfo->curTWinIdx = 0; pInfo->curTWinIdx = 0;
pInfo->lastStatus.ts = pInfo->expStatus.ts; }
pInfo->lastStatus.uid = pInfo->expStatus.uid;
return doTableScan(pOperator);
} }
// reset to exp table and window start from ts // reset to exp table and window start from ts
tsdbSetTableId(pInfo->dataReader, pInfo->expStatus.uid); tsdbSetTableId(pInfo->dataReader, pInfo->expStatus.uid);
@ -546,9 +548,23 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
tsdbResetReadHandle(pInfo->dataReader, &tmpCond, 0); tsdbResetReadHandle(pInfo->dataReader, &tmpCond, 0);
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
pInfo->curTWinIdx = 0; pInfo->curTWinIdx = 0;
pInfo->lastStatus.ts = pInfo->expStatus.ts; while (1) {
pInfo->lastStatus.uid = pInfo->expStatus.uid; SSDataBlock* result = doTableScanGroup(pOperator);
return doTableScan(pOperator); if (result) {
return result;
}
// if no data, switch to next table and continue scan
pInfo->currentTable++;
if (pInfo->currentTable >= taosArrayGetSize(pTaskInfo->tableqinfoList.pTableList)) {
return NULL;
}
STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, pInfo->currentTable);
/*pTableInfo->uid */
tsdbSetTableId(pInfo->dataReader, pTableInfo->uid);
tsdbResetReadHandle(pInfo->dataReader, &pInfo->cond, 0);
pInfo->scanTimes = 0;
pInfo->curTWinIdx = 0;
}
} }
if (pInfo->currentGroupId == -1) { if (pInfo->currentGroupId == -1) {