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

fix: invalid msg order issue

This commit is contained in:
dapan1121 2023-02-06 13:46:47 +08:00
parent a2b5162ab9
commit ebbf3af3b1
1 changed files with 2 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -263,6 +263,7 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen,
SOutputData output = {0}; SOutputData output = {0};
if (NULL == ctx->sinkHandle) { if (NULL == ctx->sinkHandle) {
pOutput->queryEnd = true;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -758,7 +759,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
} }
QW_LOCK(QW_WRITE, &ctx->lock); QW_LOCK(QW_WRITE, &ctx->lock);
if (qComplete || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) { if (atomic_load_8((int8_t*)&ctx->queryEnd) || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) {
// Note: query is not running anymore // Note: query is not running anymore
QW_SET_PHASE(ctx, QW_PHASE_POST_CQUERY); QW_SET_PHASE(ctx, QW_PHASE_POST_CQUERY);
QW_UNLOCK(QW_WRITE, &ctx->lock); QW_UNLOCK(QW_WRITE, &ctx->lock);