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

fix:error in TD-23218 & remove useless logic

This commit is contained in:
wangmm0220 2023-03-20 16:26:15 +08:00
parent f80078030c
commit 506c192bff
4 changed files with 22 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/executor/executor.h
View File

@ -198,8 +198,6 @@ int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset); void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
int64_t qStreamExtractOffsetUid(qTaskInfo_t tinfo);
SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo); SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo);
const SSchemaWrapper* qExtractSchemaFromTask(qTaskInfo_t tinfo); const SSchemaWrapper* qExtractSchemaFromTask(qTaskInfo_t tinfo);

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

@ -156,35 +156,32 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta
} }
} }
if (pDataBlock == NULL && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { // get meta
if (qStreamExtractOffsetUid(task) != 0) { 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; continue;
} }
tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);
pHandle->snapshotVer + 1); qStreamExtractOffset(task, &pRsp->rspOffset);
break; break;
} }
if (pRsp->blockNum > 0) { if (pRsp->blockNum > 0) {
tqDebug("tmqsnap task exec exited, get data"); tqDebug("tmqsnap task exec exited, get data");
break;
}
SMqMetaRsp* tmp = qStreamExtractMetaMsg(task);
if (tmp->rspOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
*pOffset = tmp->rspOffset;
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");
break;
}
qStreamExtractOffset(task, &pRsp->rspOffset); qStreamExtractOffset(task, &pRsp->rspOffset);
break;
}
}
return 0; return 0;
} }

5
source/libs/executor/src/executor.c
View File

@ -993,11 +993,6 @@ SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo) {
return &pTaskInfo->streamInfo.metaRsp; return &pTaskInfo->streamInfo.metaRsp;
} }
int64_t qStreamExtractOffsetUid(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
return pTaskInfo->streamInfo.currentOffset.uid;
}
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) { void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
memcpy(pOffset, &pTaskInfo->streamInfo.currentOffset, sizeof(STqOffsetVal)); memcpy(pOffset, &pTaskInfo->streamInfo.currentOffset, sizeof(STqOffsetVal));

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

@ -2087,17 +2087,16 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
} }
SMetaTableInfo mtInfo = getUidfromSnapShot(pInfo->sContext); SMetaTableInfo mtInfo = getUidfromSnapShot(pInfo->sContext);
STqOffsetVal offset = {0};
if (mtInfo.uid == 0) { // read snapshot done, change to get data from wal if (mtInfo.uid == 0) { // read snapshot done, change to get data from wal
qDebug("tmqsnap read snapshot done, change to get data from wal"); qDebug("tmqsnap read snapshot done, change to get data from wal");
tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pInfo->sContext->snapVersion); tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion);
} else { } else {
STqOffsetVal offset = {0};
tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN); tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN);
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid); qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);
} }
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
tDeleteSSchemaWrapper(mtInfo.schema); tDeleteSSchemaWrapper(mtInfo.schema);
qDebug("tmqsnap stream scan tsdb return null");
return NULL; return NULL;
} else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) { } else if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_META) {
SSnapContext* sContext = pInfo->sContext; SSnapContext* sContext = pInfo->sContext;
@ -2112,10 +2111,11 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
} }
if (!sContext->queryMeta) { // change to get data next poll request if (!sContext->queryMeta) { // change to get data next poll request
tqOffsetResetToData(&pTaskInfo->streamInfo.metaRsp.rspOffset, 0, INT64_MIN); STqOffsetVal offset = {0};
tqOffsetResetToData(&offset, 0, INT64_MIN);
qStreamPrepareScan(pTaskInfo, &offset, pInfo->sContext->subType);
} else { } else {
tqOffsetResetToMeta(&pTaskInfo->streamInfo.currentOffset, uid); tqOffsetResetToMeta(&pTaskInfo->streamInfo.currentOffset, uid);
pTaskInfo->streamInfo.metaRsp.rspOffset = pTaskInfo->streamInfo.currentOffset;
pTaskInfo->streamInfo.metaRsp.resMsgType = type; pTaskInfo->streamInfo.metaRsp.resMsgType = type;
pTaskInfo->streamInfo.metaRsp.metaRspLen = dataLen; pTaskInfo->streamInfo.metaRsp.metaRspLen = dataLen;
pTaskInfo->streamInfo.metaRsp.metaRsp = data; pTaskInfo->streamInfo.metaRsp.metaRsp = data;