From 0c94c65acb6e81c367f0d9364f803dbb274444e8 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 3 Jun 2022 23:13:21 +0800
Subject: [PATCH] fix(query): add ref for job

---
 source/client/src/clientImpl.c     |  2 +-
 source/libs/catalog/src/ctgAsync.c | 12 ++++++++++++
 2 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c
index 774ef5f248..4c68edaff0 100644
--- a/source/client/src/clientImpl.c
+++ b/source/client/src/clientImpl.c
@@ -1087,7 +1087,7 @@ void* doAsyncFetchRow(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUc
     tsem_wait(&pParam->sem);
   }
 
-  if (pRequest->code == TSDB_CODE_SUCCESS && setupOneRowPtr) {
+  if (pRequest->code == TSDB_CODE_SUCCESS && pResultInfo->numOfRows > 0 && setupOneRowPtr) {
     doSetOneRowPtr(pResultInfo);
     pResultInfo->current += 1;
   }
diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c
index 312f0c9250..a9e808e749 100644
--- a/source/libs/catalog/src/ctgAsync.c
+++ b/source/libs/catalog/src/ctgAsync.c
@@ -328,6 +328,18 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
     CTG_ERR_JRET(ctgInitGetQnodeTask(pJob, taskIdx++));
   }
 
+  pJob->refId = taosAddRef(gCtgMgmt.jobPool, pJob);
+  if (pJob->refId < 0) {
+    ctgError("add job to ref failed, error: %s", tstrerror(terrno));
+    CTG_ERR_JRET(terrno);
+  }
+
+  taosAcquireRef(gCtgMgmt.jobPool, pJob->refId);
+
+  qDebug("QID:%" PRIx64 ", job %" PRIx64 " initialized, task num %d", pJob->queryId, pJob->refId, *taskNum);
+  return TSDB_CODE_SUCCESS;
+
+
 _return:
   taosMemoryFreeClear(*job);
   CTG_RET(code);