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

fix:lose consume data because of exec close if consume while insert data

This commit is contained in:
wangmm0220 2023-04-03 19:54:52 +08:00
parent 1114d358a4
commit d651ba02ae
6 changed files with 12 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -198,6 +198,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
// //
int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit); int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
void qStreamSetOpen(qTaskInfo_t tinfo);
void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset); void qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo); SMqMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo);

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

@ -253,7 +253,7 @@ static void doPushDataForEntry(void* pIter, STqExecHandle* pExec, STQ* pTq, int6
if (qStreamSetScanMemData(pTaskInfo, submit) != 0) { if (qStreamSetScanMemData(pTaskInfo, submit) != 0) {
return; return;
} }
qStreamSetOpen(pTaskInfo);
// here start to scan submit block to extract the subscribed data // here start to scan submit block to extract the subscribed data
int32_t totalRows = 0; int32_t totalRows = 0;

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

@ -81,7 +81,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
while (1) { while (1) {
SSDataBlock* pDataBlock = NULL; SSDataBlock* pDataBlock = NULL;
uint64_t ts = 0; uint64_t ts = 0;
qStreamSetOpen(task);
tqDebug("consumer:0x%"PRIx64" vgId:%d, tmq one task start execute", pHandle->consumerId, vgId); tqDebug("consumer:0x%"PRIx64" vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);
if (qExecTask(task, &pDataBlock, &ts) != TSDB_CODE_SUCCESS) { if (qExecTask(task, &pDataBlock, &ts) != TSDB_CODE_SUCCESS) {
tqError("consumer:0x%"PRIx64" vgId:%d, task exec error since %s", pHandle->consumerId, vgId, terrstr()); tqError("consumer:0x%"PRIx64" vgId:%d, task exec error since %s", pHandle->consumerId, vgId, terrstr());

2
source/libs/executor/inc/executorimpl.h
View File

@ -192,7 +192,7 @@ enum {
OP_OPENED = 0x1, OP_OPENED = 0x1,
OP_RES_TO_RETURN = 0x5, OP_RES_TO_RETURN = 0x5,
OP_EXEC_DONE = 0x9, OP_EXEC_DONE = 0x9,
OP_EXEC_RECV = 0x11, // OP_EXEC_RECV = 0x11,
}; };
typedef struct SOperatorFpSet { typedef struct SOperatorFpSet {

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

@ -1074,6 +1074,12 @@ int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit) {
return 0; return 0;
} }
void qStreamSetOpen(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot;
pOperator->status = OP_NOT_OPENED;
}
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) { int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
SOperatorInfo* pOperator = pTaskInfo->pRoot; SOperatorInfo* pOperator = pTaskInfo->pRoot;
@ -1086,8 +1092,6 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
} }
if (subType == TOPIC_SUB_TYPE__COLUMN) { if (subType == TOPIC_SUB_TYPE__COLUMN) {
pOperator->status = OP_OPENED;
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
if (pOperator->numOfDownstream != 1) { if (pOperator->numOfDownstream != 1) {
qError("invalid operator, number of downstream:%d, %s", pOperator->numOfDownstream, id); qError("invalid operator, number of downstream:%d, %s", pOperator->numOfDownstream, id);

15
source/libs/executor/src/projectoperator.c
View File

@ -227,17 +227,8 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
blockDataCleanup(pFinalRes); blockDataCleanup(pFinalRes);
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (pTaskInfo->streamInfo.submit.msgStr) {
pOperator->status = OP_OPENED;
}
if (pOperator->status == OP_EXEC_DONE) { if (pOperator->status == OP_EXEC_DONE) {
// if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {
// pOperator->status = OP_OPENED;
// qDebug("projection in queue model, set status open and return null");
// return NULL;
// }
return NULL; return NULL;
} }
@ -262,11 +253,7 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// The downstream exec may change the value of the newgroup, so use a local variable instead. // The downstream exec may change the value of the newgroup, so use a local variable instead.
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream); SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL && pTaskInfo->execModel != OPTR_EXEC_MODEL_QUEUE) { if (pBlock == NULL) {
// if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE && pFinalRes->info.rows == 0) {
// pOperator->status = OP_OPENED;
// return NULL;
// }
qDebug("set op close, exec %d, status %d rows %" PRId64 , pTaskInfo->execModel, pOperator->status, pFinalRes->info.rows); qDebug("set op close, exec %d, status %d rows %" PRId64 , pTaskInfo->execModel, pOperator->status, pFinalRes->info.rows);
setOperatorCompleted(pOperator); setOperatorCompleted(pOperator);
break; break;