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

fix: return error response while fetching in quick response mode

This commit is contained in:
dapan1121 2023-11-03 11:28:16 +08:00
parent 92240d7f95
commit bb48b457ca
1 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/libs/qworker/src/qworker.c
View File

@ -434,17 +434,22 @@ int32_t qwQuickRspFetchReq(QW_FPARAMS_DEF, SQWTaskCtx * ctx, SQWMsg *qwMsg, i
void *rsp = NULL; void *rsp = NULL;
int32_t dataLen = 0; int32_t dataLen = 0;
SOutputData sOutput = {0}; SOutputData sOutput = {0};
if (qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput)) { if (TSDB_CODE_SUCCESS == code) {
code = qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput);
}
if (NULL == rsp && TSDB_CODE_SUCCESS == code) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if (rsp) { if (NULL != rsp) {
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd); bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete); qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
if (qComplete) { if (qComplete) {
atomic_store_8((int8_t *)&ctx->queryEnd, true); atomic_store_8((int8_t *)&ctx->queryEnd, true);
} }
}
qwMsg->connInfo = ctx->dataConnInfo; qwMsg->connInfo = ctx->dataConnInfo;
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH); QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
@ -456,7 +461,6 @@ int32_t qwQuickRspFetchReq(QW_FPARAMS_DEF, SQWTaskCtx * ctx, SQWMsg *qwMsg, i
dataLen); dataLen);
} }
} }
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }