From b2c8d2161fc4fea47eeb390da968defeaa468358 Mon Sep 17 00:00:00 2001
From: slzhou <shenglian_zhou@163.com>
Date: Sat, 6 Aug 2022 10:47:15 +0800
Subject: [PATCH 01/12] fix: all null value group with all null out

---
 source/libs/function/src/builtinsimpl.c | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index e78a78e7d8..7fd48519a3 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -3625,6 +3625,11 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   // todo assign the tag value and the corresponding row data
   int32_t currentRow = pBlock->info.rows;
+  if (pEntryInfo->numOfRes <= 0) {
+    colDataAppendNULL(pCol, currentRow);
+    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    return pEntryInfo->numOfRes;
+  }
   for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
     STopBotResItem* pItem = &pRes->pItems[i];
     if (type == TSDB_DATA_TYPE_FLOAT) {
@@ -4957,6 +4962,11 @@ int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   int32_t currentRow = pBlock->info.rows;
+  if (pInfo->numSampled == 0) {
+    colDataAppendNULL(pCol, currentRow);
+    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    return pInfo->numSampled;
+  }
   for (int32_t i = 0; i < pInfo->numSampled; ++i) {
     colDataAppend(pCol, currentRow + i, pInfo->data + i * pInfo->colBytes, false);
     setSelectivityValue(pCtx, pBlock, &pInfo->tuplePos[i], currentRow + i);

From 580b2e587b65833b4bb5e37e4a09fa281b5a6086 Mon Sep 17 00:00:00 2001
From: slzhou <shenglian_zhou@163.com>
Date: Sat, 6 Aug 2022 14:29:12 +0800
Subject: [PATCH 02/12] fix: set selectivity value for the null group

---
 source/libs/function/src/builtinsimpl.c | 43 ++++++++++++++++++++++---
 1 file changed, 39 insertions(+), 4 deletions(-)

diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 7fd48519a3..e3e98a6895 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -64,6 +64,9 @@ typedef struct SMinmaxResInfo {
   bool      assign;  // assign the first value or not
   int64_t   v;
   STuplePos tuplePos;
+
+  STuplePos nullTuplePos;
+  bool      nullTupleSaved;
 } SMinmaxResInfo;
 
 typedef struct STopBotResItem {
@@ -75,6 +78,10 @@ typedef struct STopBotResItem {
 typedef struct STopBotRes {
   int32_t         maxSize;
   int16_t         type;
+
+  STuplePos       nullTuplePos;
+  bool            nullTupleSaved;
+
   STopBotResItem* pItems;
 } STopBotRes;
 
@@ -221,6 +228,10 @@ typedef struct SSampleInfo {
   int32_t    numSampled;
   uint8_t    colType;
   int16_t    colBytes;
+
+  STuplePos  nullTuplePos;
+  bool       nullTupleSaved;
+
   char*      data;
   STuplePos* tuplePos;
 } SSampleInfo;
@@ -1134,6 +1145,9 @@ bool minmaxFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo)
   SMinmaxResInfo* buf = GET_ROWCELL_INTERBUF(pResultInfo);
   buf->assign = false;
   buf->tuplePos.pageId = -1;
+
+  buf->nullTupleSaved = false;
+  buf->nullTuplePos.pageId = -1;
   return true;
 }
 
@@ -1575,6 +1589,10 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
   }
 
 _min_max_over:
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pBuf->nullTupleSaved ) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pBuf->nullTuplePos);
+    pBuf->nullTupleSaved = true;
+  }
   return numOfElems;
 }
 
@@ -1615,7 +1633,7 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   if (pEntryInfo->numOfRes > 0) {
     setSelectivityValue(pCtx, pBlock, &pRes->tuplePos, currentRow);
   } else {
-    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
   }
 
   return pEntryInfo->numOfRes;
@@ -3366,6 +3384,8 @@ bool topBotFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
 
   pRes->maxSize = pCtx->param[1].param.i;
 
+  pRes->nullTupleSaved = false;
+  pRes->nullTuplePos.pageId = -1;
   return true;
 }
 
@@ -3403,6 +3423,10 @@ int32_t topFunction(SqlFunctionCtx* pCtx) {
     doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, true);
   }
 
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    pRes->nullTupleSaved = true;
+  }
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3427,6 +3451,11 @@ int32_t bottomFunction(SqlFunctionCtx* pCtx) {
     doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, pRes->type, pInput->uid, pResInfo, false);
   }
 
+  if (numOfElems == 0 && pCtx->subsidiaries.num > 0 && !pRes->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pRes->nullTuplePos);
+    pRes->nullTupleSaved = true;
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3627,7 +3656,7 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   int32_t currentRow = pBlock->info.rows;
   if (pEntryInfo->numOfRes <= 0) {
     colDataAppendNULL(pCol, currentRow);
-    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pRes->nullTuplePos, currentRow);
     return pEntryInfo->numOfRes;
   }
   for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
@@ -4902,7 +4931,8 @@ bool sampleFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo)
   pInfo->numSampled = 0;
   pInfo->colType = pCtx->resDataInfo.type;
   pInfo->colBytes = pCtx->resDataInfo.bytes;
-
+  pInfo->nullTuplePos.pageId = -1;
+  pInfo->nullTupleSaved = false;
   pInfo->data = (char*)pInfo + sizeof(SSampleInfo);
   pInfo->tuplePos = (STuplePos*)((char*)pInfo + sizeof(SSampleInfo) + pInfo->samples * pInfo->colBytes);
 
@@ -4948,6 +4978,11 @@ int32_t sampleFunction(SqlFunctionCtx* pCtx) {
     doReservoirSample(pCtx, pInfo, data, i);
   }
 
+  if (pInfo->numSampled == 0 && pCtx->subsidiaries.num > 0 && !pInfo->nullTupleSaved) {
+    doSaveTupleData(pCtx, pInput->startRowIndex, pCtx->pSrcBlock, &pInfo->nullTuplePos);
+    pInfo->nullTupleSaved = true;
+  }
+
   SET_VAL(pResInfo, pInfo->numSampled, pInfo->numSampled);
   return TSDB_CODE_SUCCESS;
 }
@@ -4964,7 +4999,7 @@ int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   int32_t currentRow = pBlock->info.rows;
   if (pInfo->numSampled == 0) {
     colDataAppendNULL(pCol, currentRow);
-    setNullSelectivityValue(pCtx, pBlock, currentRow);
+    setSelectivityValue(pCtx, pBlock, &pInfo->nullTuplePos, currentRow);
     return pInfo->numSampled;
   }
   for (int32_t i = 0; i < pInfo->numSampled; ++i) {

From ecd560ccb0ae7ecd8d93654a90ad6427c0994c0e Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 6 Aug 2022 15:30:11 +0800
Subject: [PATCH 03/12] start timer for particular msg

---
 include/libs/transport/trpc.h            | 21 +++++---
 include/util/taoserror.h                 |  1 +
 source/client/src/clientEnv.c            | 17 +++++--
 source/libs/transport/inc/transComm.h    |  5 +-
 source/libs/transport/inc/transportInt.h |  1 +
 source/libs/transport/src/trans.c        |  1 +
 source/libs/transport/src/transCli.c     | 65 ++++++++++++++----------
 source/os/src/osSocket.c                 |  6 +--
 source/util/src/terror.c                 |  1 +
 9 files changed, 73 insertions(+), 45 deletions(-)

diff --git a/include/libs/transport/trpc.h b/include/libs/transport/trpc.h
index 50f9959177..467f0a9ff0 100644
--- a/include/libs/transport/trpc.h
+++ b/include/libs/transport/trpc.h
@@ -41,12 +41,13 @@ typedef struct {
 
 typedef struct SRpcHandleInfo {
   // rpc info
-  void    *handle;         // rpc handle returned to app
-  int64_t  refId;          // refid, used by server
-  int32_t  noResp;         // has response or not(default 0, 0: resp, 1: no resp);
-  int32_t  persistHandle;  // persist handle or not
+  void   *handle;         // rpc handle returned to app
+  int64_t refId;          // refid, used by server
+  int8_t  noResp;         // has response or not(default 0, 0: resp, 1: no resp)
+  int8_t  persistHandle;  // persist handle or not
+  int8_t  hasEpSet;
+
   STraceId traceId;
-  int8_t   hasEpSet;
 
   // app info
   void *ahandle;  // app handle set by client
@@ -69,8 +70,9 @@ typedef struct SRpcMsg {
   SRpcHandleInfo info;
 } SRpcMsg;
 
-typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *rf);
+typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *epset);
 typedef bool (*RpcRfp)(int32_t code, tmsg_t msgType);
+typedef bool (*RpcTfp)(int32_t code, tmsg_t msgType);
 
 typedef struct SRpcInit {
   char     localFqdn[TSDB_FQDN_LEN];
@@ -84,12 +86,15 @@ typedef struct SRpcInit {
   // the following is for client app ecurity only
   char *user;  // user name
 
-  // call back to process incoming msg, code shall be ignored by server app
+  // call back to process incoming msg
   RpcCfp cfp;
 
-  // user defined retry func
+  // retry not not for particular msg
   RpcRfp rfp;
 
+  // set up timeout for particular msg
+  RpcTfp tfp;
+
   void *parent;
 } SRpcInit;
 
diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 97a664d776..27fb057b44 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -46,6 +46,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RPC_FQDN_ERROR                TAOS_DEF_ERROR_CODE(0, 0x0015)
 #define TSDB_CODE_RPC_PORT_EADDRINUSE           TAOS_DEF_ERROR_CODE(0, 0x0017)
 #define TSDB_CODE_RPC_BROKEN_LINK               TAOS_DEF_ERROR_CODE(0, 0x0018)
+#define TSDB_CODE_RPC_TIMEOUT                   TAOS_DEF_ERROR_CODE(0, 0x0019)
 
 //common & util
 #define TSDB_CODE_TIME_UNSYNCED                 TAOS_DEF_ERROR_CODE(0, 0x0013) 
diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c
index d88a587f6a..ae80868167 100644
--- a/source/client/src/clientEnv.c
+++ b/source/client/src/clientEnv.c
@@ -60,7 +60,7 @@ static int32_t registerRequest(SRequestObj *pRequest, STscObj *pTscObj) {
 }
 
 static void deregisterRequest(SRequestObj *pRequest) {
-  const static int64_t SLOW_QUERY_INTERVAL = 3000000L; // todo configurable
+  const static int64_t SLOW_QUERY_INTERVAL = 3000000L;  // todo configurable
   assert(pRequest != NULL);
 
   STscObj            *pTscObj = pRequest->pTscObj;
@@ -77,13 +77,13 @@ static void deregisterRequest(SRequestObj *pRequest) {
   if (QUERY_NODE_VNODE_MODIF_STMT == pRequest->stmtType) {
     atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
   } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
-    atomic_add_fetch_64((int64_t *)&pActivity->queryElapsedTime, duration);           
+    atomic_add_fetch_64((int64_t *)&pActivity->queryElapsedTime, duration);
   }
-  
+
   if (duration >= SLOW_QUERY_INTERVAL) {
     atomic_add_fetch_64((int64_t *)&pActivity->numOfSlowQueries, 1);
   }
-  
+
   releaseTscObj(pTscObj->id);
 }
 
@@ -109,6 +109,12 @@ static bool clientRpcRfp(int32_t code, tmsg_t msgType) {
   }
 }
 
+// start timer for particular msgType
+static bool clientRpcTfp(int32_t code, tmsg_t msgType) {
+  //
+  return false;
+}
+
 // TODO refactor
 void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   SRpcInit rpcInit;
@@ -118,6 +124,7 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.numOfThreads = numOfThread;
   rpcInit.cfp = processMsgFromServer;
   rpcInit.rfp = clientRpcRfp;
+  rpcInit.tfp = clientRpcTfp;
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.user = (char *)user;
@@ -375,7 +382,7 @@ void taos_init_imp(void) {
   initQueryModuleMsgHandle();
 
   taosConvInit();
-  
+
   rpcInit();
 
   SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h
index a81d6db80f..44075f316c 100644
--- a/source/libs/transport/inc/transComm.h
+++ b/source/libs/transport/inc/transComm.h
@@ -95,8 +95,9 @@ typedef void* queue[2];
 #define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
 
 #define TRANS_RETRY_COUNT_LIMIT 100  // retry count limit
-#define TRANS_RETRY_INTERVAL    15   // ms retry interval
-#define TRANS_CONN_TIMEOUT      3    // connect timeout
+#define TRANS_RETRY_INTERVAL    15   // retry interval (ms)
+#define TRANS_CONN_TIMEOUT      3    // connect timeout (s)
+#define TRANS_READ_TIMEOUT      200  // read timeout  (ms)
 
 typedef SRpcMsg      STransMsg;
 typedef SRpcCtx      STransCtx;
diff --git a/source/libs/transport/inc/transportInt.h b/source/libs/transport/inc/transportInt.h
index 5fb2980ceb..6aeeffa192 100644
--- a/source/libs/transport/inc/transportInt.h
+++ b/source/libs/transport/inc/transportInt.h
@@ -53,6 +53,7 @@ typedef struct {
 
   void (*cfp)(void* parent, SRpcMsg*, SEpSet*);
   bool (*retry)(int32_t code, tmsg_t msgType);
+  bool (*startTimer)(int32_t code, tmsg_t msgType);
   int index;
 
   void*         parent;
diff --git a/source/libs/transport/src/trans.c b/source/libs/transport/src/trans.c
index 7633820292..0a0dcef378 100644
--- a/source/libs/transport/src/trans.c
+++ b/source/libs/transport/src/trans.c
@@ -48,6 +48,7 @@ void* rpcOpen(const SRpcInit* pInit) {
   // register callback handle
   pRpc->cfp = pInit->cfp;
   pRpc->retry = pInit->rfp;
+  pRpc->startTimer = pInit->tfp;
 
   if (pInit->connType == TAOS_CONN_SERVER) {
     pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c
index 431e479123..6458fc99b7 100644
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@@ -24,6 +24,7 @@ typedef struct SCliConn {
   uv_connect_t connReq;
   uv_stream_t* stream;
   queue        wreqQueue;
+  uv_timer_t*  timer;
 
   void* hostThrd;
 
@@ -108,6 +109,8 @@ static int sockDebugInfo(struct sockaddr* sockname, char* dst) {
   sprintf(dst, "%s:%d", buf, ntohs(addr.sin_port));
   return r;
 }
+// register timer for read
+static void cliReadTimeoutCb(uv_timer_t* handle);
 // register timer in each thread to clear expire conn
 // static void cliTimeoutCb(uv_timer_t* handle);
 // alloc buf for recv
@@ -330,6 +333,11 @@ void cliHandleResp(SCliConn* conn) {
   SCliThrd* pThrd = conn->hostThrd;
   STrans*   pTransInst = pThrd->pTransInst;
 
+  if (uv_is_active((uv_handle_t*)conn->timer)) {
+    tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn);
+    uv_timer_stop(conn->timer);
+  }
+
   STransMsgHead* pHead = NULL;
   transDumpFromBuffer(&conn->readBuf, (char**)&pHead);
   pHead->code = htonl(pHead->code);
@@ -409,7 +417,7 @@ void cliHandleResp(SCliConn* conn) {
   uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
 }
 
-void cliHandleExcept(SCliConn* pConn) {
+void cliHandleExceptImpl(SCliConn* pConn, int32_t code) {
   if (transQueueEmpty(&pConn->cliMsgs)) {
     if (pConn->broken == true && CONN_NO_PERSIST_BY_APP(pConn)) {
       tTrace("%s conn %p handle except, persist:0", CONN_GET_INST_LABEL(pConn), pConn);
@@ -428,7 +436,7 @@ void cliHandleExcept(SCliConn* pConn) {
     STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
 
     STransMsg transMsg = {0};
-    transMsg.code = pConn->broken ? TSDB_CODE_RPC_BROKEN_LINK : TSDB_CODE_RPC_NETWORK_UNAVAIL;
+    transMsg.code = code == -1 ? (pConn->broken ? TSDB_CODE_RPC_BROKEN_LINK : TSDB_CODE_RPC_NETWORK_UNAVAIL) : code;
     transMsg.msgType = pMsg ? pMsg->msg.msgType + 1 : 0;
     transMsg.info.ahandle = NULL;
 
@@ -459,31 +467,17 @@ void cliHandleExcept(SCliConn* pConn) {
   } while (!transQueueEmpty(&pConn->cliMsgs));
   transUnrefCliHandle(pConn);
 }
+void cliHandleExcept(SCliConn* conn) {
+  tTrace("%s conn except ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
+  cliHandleExceptImpl(conn, -1);
+}
 
-// void cliTimeoutCb(uv_timer_t* handle) {
-//   SCliThrd* pThrd = handle->data;
-//   STrans*   pTransInst = pThrd->pTransInst;
-//   int64_t   currentTime = pThrd->nextTimeout;
-//   tTrace("%s conn timeout, try to remove expire conn from conn pool", pTransInst->label);
-//
-//   SConnList* p = taosHashIterate((SHashObj*)pThrd->pool, NULL);
-//   while (p != NULL) {
-//     while (!QUEUE_IS_EMPTY(&p->conn)) {
-//       queue*    h = QUEUE_HEAD(&p->conn);
-//       SCliConn* c = QUEUE_DATA(h, SCliConn, q);
-//       if (c->expireTime < currentTime) {
-//         QUEUE_REMOVE(h);
-//         transUnrefCliHandle(c);
-//       } else {
-//         break;
-//       }
-//     }
-//     p = taosHashIterate((SHashObj*)pThrd->pool, p);
-//   }
-//
-//   pThrd->nextTimeout = taosGetTimestampMs() + CONN_PERSIST_TIME(pTransInst->idleTime);
-//   uv_timer_start(handle, cliTimeoutCb, CONN_PERSIST_TIME(pTransInst->idleTime) / 2, 0);
-// }
+void cliReadTimeoutCb(uv_timer_t* handle) {
+  // set up timeout cb
+  SCliConn* conn = handle->data;
+  tTrace("%s conn %p timeout, ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
+  cliHandleExceptImpl(conn, TSDB_CODE_RPC_TIMEOUT);
+}
 
 void* createConnPool(int size) {
   // thread local, no lock
@@ -638,6 +632,11 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
 
   conn->connReq.data = conn;
 
+  // set read timeout
+  conn->timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
+  uv_timer_init(pThrd->loop, conn->timer);
+  conn->timer->data = conn;
+
   transReqQueueInit(&conn->wreqQueue);
 
   transQueueInit(&conn->cliMsgs, NULL);
@@ -660,7 +659,6 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
   transRemoveExHandle(transGetRefMgt(), conn->refId);
 
   conn->refId = -1;
-  if (conn->task != NULL) transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task);
 
   if (clear) {
     if (!uv_is_closing((uv_handle_t*)conn->stream)) {
@@ -675,6 +673,15 @@ static void cliDestroy(uv_handle_t* handle) {
   }
 
   SCliConn* conn = handle->data;
+  if (conn->timer != NULL) {
+    uv_timer_stop(conn->timer);
+    taosMemoryFree(conn->timer);
+    conn->timer = NULL;
+  }
+  if (conn->task != NULL) {
+    transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task);
+    conn->task = NULL;
+  }
   transRemoveExHandle(transGetRefMgt(), conn->refId);
   taosMemoryFree(conn->ip);
   conn->stream->data = NULL;
@@ -764,6 +771,10 @@ void cliSend(SCliConn* pConn) {
     CONN_SET_PERSIST_BY_APP(pConn);
   }
 
+  if (pTransInst->startTimer != NULL && pTransInst->startTimer(0, pMsg->msgType)) {
+    tGTrace("%s conn %p start timer for msg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pMsg->msgType));
+    uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0);
+  }
   uv_write_t* req = transReqQueuePush(&pConn->wreqQueue);
   uv_write(req, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
   return;
diff --git a/source/os/src/osSocket.c b/source/os/src/osSocket.c
index 98dfaa4972..f34032056c 100644
--- a/source/os/src/osSocket.c
+++ b/source/os/src/osSocket.c
@@ -333,7 +333,7 @@ int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
     return -1;
   }
   int32_t nleft, nwritten;
-  char *  ptr = (char *)buf;
+  char   *ptr = (char *)buf;
 
   nleft = nbytes;
 
@@ -362,7 +362,7 @@ int32_t taosReadMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
     return -1;
   }
   int32_t nleft, nread;
-  char *  ptr = (char *)buf;
+  char   *ptr = (char *)buf;
 
   nleft = nbytes;
 
@@ -912,7 +912,7 @@ uint32_t taosGetIpv4FromFqdn(const char *fqdn) {
 
   int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);
   if (result) {
-    struct sockaddr *   sa = result->ai_addr;
+    struct sockaddr    *sa = result->ai_addr;
     struct sockaddr_in *si = (struct sockaddr_in *)sa;
     struct in_addr      ia = si->sin_addr;
     uint32_t            ip = ia.s_addr;
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index e4b6983dcb..4780d85a30 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -52,6 +52,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RPC_NETWORK_UNAVAIL,          "Unable to establish c
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_FQDN_ERROR,               "Unable to resolve FQDN")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_PORT_EADDRINUSE,          "Port already in use")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_BROKEN_LINK,              "Conn is broken")
+TAOS_DEFINE_ERROR(TSDB_CODE_RPC_TIMEOUT,                  "Conn read timeout")
 
 //common & util
 TAOS_DEFINE_ERROR(TSDB_CODE_TIME_UNSYNCED,                "Client and server's time is not synchronized")

From 66adfa9e42719cd9c2ef6b12b6161d4536ace70f Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 6 Aug 2022 15:50:12 +0800
Subject: [PATCH 04/12] start timer for particular msg

---
 source/client/src/clientEnv.c         | 4 +++-
 source/libs/transport/inc/transComm.h | 8 ++++----
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c
index ae80868167..812351e208 100644
--- a/source/client/src/clientEnv.c
+++ b/source/client/src/clientEnv.c
@@ -111,7 +111,9 @@ static bool clientRpcRfp(int32_t code, tmsg_t msgType) {
 
 // start timer for particular msgType
 static bool clientRpcTfp(int32_t code, tmsg_t msgType) {
-  //
+  if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_CREATE_TABLE) {
+    return true;
+  }
   return false;
 }
 
diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h
index 44075f316c..ad0ce4f5e1 100644
--- a/source/libs/transport/inc/transComm.h
+++ b/source/libs/transport/inc/transComm.h
@@ -94,10 +94,10 @@ typedef void* queue[2];
 /* Return the structure holding the given element. */
 #define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
 
-#define TRANS_RETRY_COUNT_LIMIT 100  // retry count limit
-#define TRANS_RETRY_INTERVAL    15   // retry interval (ms)
-#define TRANS_CONN_TIMEOUT      3    // connect timeout (s)
-#define TRANS_READ_TIMEOUT      200  // read timeout  (ms)
+#define TRANS_RETRY_COUNT_LIMIT 100   // retry count limit
+#define TRANS_RETRY_INTERVAL    15    // retry interval (ms)
+#define TRANS_CONN_TIMEOUT      3     // connect timeout (s)
+#define TRANS_READ_TIMEOUT      3000  // read timeout  (ms)
 
 typedef SRpcMsg      STransMsg;
 typedef SRpcCtx      STransCtx;

From 0753087e88fedb9de6776073c99916c9154a021c Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 6 Aug 2022 17:59:30 +0800
Subject: [PATCH 05/12] start timer for particular msg

---
 source/libs/transport/src/transCli.c | 112 +++++++++++++--------------
 1 file changed, 53 insertions(+), 59 deletions(-)

diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c
index 6458fc99b7..b8346370e9 100644
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@@ -66,12 +66,13 @@ typedef struct SCliThrd {
   int64_t       pid;     // pid
   uv_loop_t*    loop;
   SAsyncPool*   asyncPool;
-  uv_idle_t*    idle;
   uv_prepare_t* prepare;
-  uv_timer_t    timer;
   void*         pool;  // conn pool
 
+  SArray* timerList;
+
   // msg queue
+
   queue         msg;
   TdThreadMutex msgMtx;
   SDelayQueue*  delayQueue;
@@ -333,9 +334,14 @@ void cliHandleResp(SCliConn* conn) {
   SCliThrd* pThrd = conn->hostThrd;
   STrans*   pTransInst = pThrd->pTransInst;
 
-  if (uv_is_active((uv_handle_t*)conn->timer)) {
-    tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn);
-    uv_timer_stop(conn->timer);
+  if (conn->timer) {
+    if (uv_is_active((uv_handle_t*)conn->timer)) {
+      tDebug("%s conn %p stop timer", CONN_GET_INST_LABEL(conn), conn);
+      uv_timer_stop(conn->timer);
+    }
+    conn->timer->data = NULL;
+    taosArrayPush(pThrd->timerList, &conn->timer);
+    conn->timer = NULL;
   }
 
   STransMsgHead* pHead = NULL;
@@ -468,7 +474,7 @@ void cliHandleExceptImpl(SCliConn* pConn, int32_t code) {
   transUnrefCliHandle(pConn);
 }
 void cliHandleExcept(SCliConn* conn) {
-  tTrace("%s conn except ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
+  tTrace("%s conn %p except ref:%d", CONN_GET_INST_LABEL(conn), conn, T_REF_VAL_GET(conn));
   cliHandleExceptImpl(conn, -1);
 }
 
@@ -632,11 +638,6 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
 
   conn->connReq.data = conn;
 
-  // set read timeout
-  conn->timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
-  uv_timer_init(pThrd->loop, conn->timer);
-  conn->timer->data = conn;
-
   transReqQueueInit(&conn->wreqQueue);
 
   transQueueInit(&conn->cliMsgs, NULL);
@@ -653,13 +654,24 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
   return conn;
 }
 static void cliDestroyConn(SCliConn* conn, bool clear) {
+  SCliThrd* pThrd = conn->hostThrd;
   tTrace("%s conn %p remove from conn pool", CONN_GET_INST_LABEL(conn), conn);
   QUEUE_REMOVE(&conn->q);
   QUEUE_INIT(&conn->q);
   transRemoveExHandle(transGetRefMgt(), conn->refId);
-
   conn->refId = -1;
 
+  if (conn->task != NULL) {
+    transDQCancel(pThrd->timeoutQueue, conn->task);
+    conn->task = NULL;
+  }
+  if (conn->timer != NULL) {
+    uv_timer_stop(conn->timer);
+    taosArrayPush(pThrd->timerList, &conn->timer);
+    conn->timer->data = NULL;
+    conn->timer = NULL;
+  }
+
   if (clear) {
     if (!uv_is_closing((uv_handle_t*)conn->stream)) {
       uv_read_stop(conn->stream);
@@ -671,17 +683,15 @@ static void cliDestroy(uv_handle_t* handle) {
   if (uv_handle_get_type(handle) != UV_TCP || handle->data == NULL) {
     return;
   }
-
   SCliConn* conn = handle->data;
+  SCliThrd* pThrd = conn->hostThrd;
   if (conn->timer != NULL) {
     uv_timer_stop(conn->timer);
-    taosMemoryFree(conn->timer);
+    taosArrayPush(pThrd->timerList, &conn->timer);
+    conn->timer->data = NULL;
     conn->timer = NULL;
   }
-  if (conn->task != NULL) {
-    transDQCancel(((SCliThrd*)conn->hostThrd)->timeoutQueue, conn->task);
-    conn->task = NULL;
-  }
+
   transRemoveExHandle(transGetRefMgt(), conn->refId);
   taosMemoryFree(conn->ip);
   conn->stream->data = NULL;
@@ -772,6 +782,15 @@ void cliSend(SCliConn* pConn) {
   }
 
   if (pTransInst->startTimer != NULL && pTransInst->startTimer(0, pMsg->msgType)) {
+    uv_timer_t* timer = taosArrayPop(pThrd->timerList);
+    if (timer == NULL) {
+      tDebug("no avaiable timer, create");
+      timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
+      uv_timer_init(pThrd->loop, timer);
+    }
+    timer->data = pConn;
+    pConn->timer = timer;
+
     tGTrace("%s conn %p start timer for msg:%s", CONN_GET_INST_LABEL(pConn), pConn, TMSG_INFO(pMsg->msgType));
     uv_timer_start((uv_timer_t*)pConn->timer, cliReadTimeoutCb, TRANS_READ_TIMEOUT, 0);
   }
@@ -792,8 +811,8 @@ void cliConnCb(uv_connect_t* req, int status) {
   }
   // int addrlen = sizeof(pConn->addr);
   struct sockaddr peername, sockname;
-  int             addrlen = sizeof(peername);
 
+  int addrlen = sizeof(peername);
   uv_tcp_getpeername((uv_tcp_t*)pConn->stream, &peername, &addrlen);
   transGetSockDebugInfo(&peername, pConn->dst);
 
@@ -817,7 +836,6 @@ static void cliHandleQuit(SCliMsg* pMsg, SCliThrd* pThrd) {
   tDebug("cli work thread %p start to quit", pThrd);
   destroyCmsg(pMsg);
   destroyConnPool(pThrd->pool);
-  uv_timer_stop(&pThrd->timer);
   uv_walk(pThrd->loop, cliWalkCb, NULL);
 }
 static void cliHandleRelease(SCliMsg* pMsg, SCliThrd* pThrd) {
@@ -890,8 +908,8 @@ void cliMayCvtFqdnToIp(SEpSet* pEpSet, SCvtAddr* pCvtAddr) {
   }
 }
 void cliHandleReq(SCliMsg* pMsg, SCliThrd* pThrd) {
-  STransConnCtx* pCtx = pMsg->ctx;
   STrans*        pTransInst = pThrd->pTransInst;
+  STransConnCtx* pCtx = pMsg->ctx;
 
   cliMayCvtFqdnToIp(&pCtx->epSet, &pThrd->cvtAddr);
   if (!EPSET_IS_VALID(&pCtx->epSet)) {
@@ -977,36 +995,6 @@ static void cliAsyncCb(uv_async_t* handle) {
   }
   if (pThrd->stopMsg != NULL) cliHandleQuit(pThrd->stopMsg, pThrd);
 }
-static void cliIdleCb(uv_idle_t* handle) {
-  SCliThrd* thrd = handle->data;
-  tTrace("do idle work");
-
-  SAsyncPool* pool = thrd->asyncPool;
-  for (int i = 0; i < pool->nAsync; i++) {
-    uv_async_t* async = &(pool->asyncs[i]);
-    SAsyncItem* item = async->data;
-
-    queue wq;
-    taosThreadMutexLock(&item->mtx);
-    QUEUE_MOVE(&item->qmsg, &wq);
-    taosThreadMutexUnlock(&item->mtx);
-
-    int count = 0;
-    while (!QUEUE_IS_EMPTY(&wq)) {
-      queue* h = QUEUE_HEAD(&wq);
-      QUEUE_REMOVE(h);
-
-      SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
-      if (pMsg == NULL) {
-        continue;
-      }
-      (*cliAsyncHandle[pMsg->type])(pMsg, thrd);
-      count++;
-    }
-  }
-  tTrace("prepare work end");
-  if (thrd->stopMsg != NULL) cliHandleQuit(thrd->stopMsg, thrd);
-}
 static void cliPrepareCb(uv_prepare_t* handle) {
   SCliThrd* thrd = handle->data;
   tTrace("prepare work start");
@@ -1096,19 +1084,20 @@ static SCliThrd* createThrdObj() {
   uv_loop_init(pThrd->loop);
 
   pThrd->asyncPool = transAsyncPoolCreate(pThrd->loop, 5, pThrd, cliAsyncCb);
-  uv_timer_init(pThrd->loop, &pThrd->timer);
-  pThrd->timer.data = pThrd;
-
-  // pThrd->idle = taosMemoryCalloc(1, sizeof(uv_idle_t));
-  // uv_idle_init(pThrd->loop, pThrd->idle);
-  // pThrd->idle->data = pThrd;
-  // uv_idle_start(pThrd->idle, cliIdleCb);
 
   pThrd->prepare = taosMemoryCalloc(1, sizeof(uv_prepare_t));
   uv_prepare_init(pThrd->loop, pThrd->prepare);
   pThrd->prepare->data = pThrd;
   uv_prepare_start(pThrd->prepare, cliPrepareCb);
 
+  int32_t timerSize = 512;
+  pThrd->timerList = taosArrayInit(timerSize, sizeof(void*));
+  for (int i = 0; i < timerSize; i++) {
+    uv_timer_t* timer = taosMemoryCalloc(1, sizeof(uv_timer_t));
+    uv_timer_init(pThrd->loop, timer);
+    taosArrayPush(pThrd->timerList, &timer);
+  }
+
   pThrd->pool = createConnPool(4);
   transDQCreate(pThrd->loop, &pThrd->delayQueue);
 
@@ -1131,7 +1120,12 @@ static void destroyThrdObj(SCliThrd* pThrd) {
   transDQDestroy(pThrd->delayQueue, destroyCmsg);
   transDQDestroy(pThrd->timeoutQueue, NULL);
 
-  taosMemoryFree(pThrd->idle);
+  for (int i = 0; i < taosArrayGetSize(pThrd->timerList); i++) {
+    uv_timer_t* timer = taosArrayGetP(pThrd->timerList, i);
+    taosMemoryFree(timer);
+  }
+  taosArrayDestroy(pThrd->timerList);
+
   taosMemoryFree(pThrd->prepare);
   taosMemoryFree(pThrd->loop);
   taosMemoryFree(pThrd);

From fd654b126a16d32ae9d78d4078cb4a03cf885e9a Mon Sep 17 00:00:00 2001
From: dapan1121 <wpan@taosdata.com>
Date: Sat, 6 Aug 2022 18:00:26 +0800
Subject: [PATCH 06/12] fix: fix taosd memory leak issue

---
 include/libs/executor/executor.h        | 2 +-
 source/libs/executor/inc/executorimpl.h | 2 +-
 source/libs/executor/src/executor.c     | 2 +-
 source/libs/executor/src/executorimpl.c | 4 +++-
 source/libs/qworker/inc/qwMsg.h         | 2 +-
 source/libs/qworker/src/qworker.c       | 5 ++++-
 6 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h
index ab33af6acf..7bc2ede4fc 100644
--- a/include/libs/executor/executor.h
+++ b/include/libs/executor/executor.h
@@ -103,7 +103,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
  * @return
  */
 int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan,
-                        qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, const char* sql, EOPTR_EXEC_MODEL model);
+                        qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, char* sql, EOPTR_EXEC_MODEL model);
 
 /**
  *
diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index b62ff2bef1..631d88c684 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -987,7 +987,7 @@ int32_t decodeOperator(SOperatorInfo* ops, const char* data, int32_t length);
 
 void    setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
 int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
-                               const char* sql, EOPTR_EXEC_MODEL model);
+                               char* sql, EOPTR_EXEC_MODEL model);
 int32_t createDataSinkParam(SDataSinkNode *pNode, void **pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
 int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
 
diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index f080549578..897ddb59ec 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -341,7 +341,7 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table
 }
 
 int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan,
-                        qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, const char* sql, EOPTR_EXEC_MODEL model) {
+                        qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, char* sql, EOPTR_EXEC_MODEL model) {
   assert(pSubplan != NULL);
   SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
 
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index eaffbc2579..3a61661c5a 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -4492,7 +4492,7 @@ int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pT
 }
 
 int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
-                               const char* sql, EOPTR_EXEC_MODEL model) {
+                               char* sql, EOPTR_EXEC_MODEL model) {
   uint64_t queryId = pPlan->id.queryId;
 
   int32_t code = TSDB_CODE_SUCCESS;
@@ -4503,6 +4503,7 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
   }
 
   (*pTaskInfo)->sql = sql;
+  sql = NULL;
   (*pTaskInfo)->pSubplan = pPlan;
   (*pTaskInfo)->pRoot = createOperatorTree(pPlan->pNode, *pTaskInfo, pHandle, &(*pTaskInfo)->tableqinfoList,
                                            pPlan->pTagCond, pPlan->pTagIndexCond, pPlan->user);
@@ -4515,6 +4516,7 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
   return code;
 
 _complete:
+  taosMemoryFree(sql);
   doDestroyTask(*pTaskInfo);
   terrno = code;
   return code;
diff --git a/source/libs/qworker/inc/qwMsg.h b/source/libs/qworker/inc/qwMsg.h
index 3ee870ef96..3ff5b5950f 100644
--- a/source/libs/qworker/inc/qwMsg.h
+++ b/source/libs/qworker/inc/qwMsg.h
@@ -25,7 +25,7 @@ extern "C" {
 
 int32_t qwAbortPrerocessQuery(QW_FPARAMS_DEF);
 int32_t qwPreprocessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg);
-int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, const char* sql);
+int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char* sql);
 int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg);
 int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg);
 int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg);
diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c
index 36d85f1f12..04fca6f0d9 100644
--- a/source/libs/qworker/src/qworker.c
+++ b/source/libs/qworker/src/qworker.c
@@ -508,7 +508,7 @@ _return:
 }
 
 
-int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, const char* sql) {
+int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char* sql) {
   int32_t        code = 0;
   bool           queryRsped = false;
   SSubplan      *plan = NULL;
@@ -536,6 +536,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, const char* sql) {
   }
 
   code = qCreateExecTask(qwMsg->node, mgmt->nodeId, tId, plan, &pTaskInfo, &sinkHandle, sql, OPTR_EXEC_MODEL_BATCH);
+  sql = NULL;
   if (code) {
     QW_TASK_ELOG("qCreateExecTask failed, code:%x - %s", code, tstrerror(code));
     QW_ERR_JRET(code);
@@ -561,6 +562,8 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, const char* sql) {
 
 _return:
 
+  taosMemoryFree(sql);
+  
   input.code = code;
   input.msgType = qwMsg->msgType;
   code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL);

From 3a3c6fac6bee1e454deecdbc8b94aa2f4ea1336a Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Sat, 6 Aug 2022 18:08:46 +0800
Subject: [PATCH 07/12] start timer for particular msg

---
 source/libs/transport/src/transCli.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c
index b8346370e9..dc7ffa9b13 100644
--- a/source/libs/transport/src/transCli.c
+++ b/source/libs/transport/src/transCli.c
@@ -782,7 +782,7 @@ void cliSend(SCliConn* pConn) {
   }
 
   if (pTransInst->startTimer != NULL && pTransInst->startTimer(0, pMsg->msgType)) {
-    uv_timer_t* timer = taosArrayPop(pThrd->timerList);
+    uv_timer_t* timer = taosArrayGetSize(pThrd->timerList) > 0 ? *(uv_timer_t**)taosArrayPop(pThrd->timerList) : NULL;
     if (timer == NULL) {
       tDebug("no avaiable timer, create");
       timer = taosMemoryCalloc(1, sizeof(uv_timer_t));

From d556d9290edefb39d43300b0b854fe49e16cee91 Mon Sep 17 00:00:00 2001
From: Shuduo Sang <sangshuduo@gmail.com>
Date: Sat, 6 Aug 2022 19:19:26 +0800
Subject: [PATCH 08/12] feat: update taostools 3c7dafe for3.0 (#15816)

* feat: update taos-tools

for 3.0

[TD-14141]

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools for 3.0

* feat: update taos-tools 8e3b3ee

* fix: remove submodules

* feat: update taos-tools c529299

* feat: update taos-tools 9dc2fec for 3.0

* fix: optim upx

* feat: update taos-tools f4e456a for 3.0

* feat: update taos-tools 2a2def1 for 3.0

* feat: update taos-tools c9cc20f for 3.0

* feat: update taostoosl 8a5e336 for 3.0

* feat: update taostools 3c7dafe for 3.0
---
 cmake/taostools_CMakeLists.txt.in | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in
index 620faa636b..9ee0ea526c 100644
--- a/cmake/taostools_CMakeLists.txt.in
+++ b/cmake/taostools_CMakeLists.txt.in
@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 8a5e336
+        GIT_TAG 3c7dafe
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

From a16a6b73c8ff73976daa63a667a339184113e4d1 Mon Sep 17 00:00:00 2001
From: shenglian zhou <slzhou@taosdata.com>
Date: Sat, 6 Aug 2022 20:11:12 +0800
Subject: [PATCH 09/12] fix: force at least one row for agg/window group

---
 source/libs/executor/src/executorimpl.c | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 7c19ad024e..b99efce106 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -1448,6 +1448,10 @@ static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_
       pRow->numOfRows = pResInfo->numOfRes;
     }
   }
+  // TODO: if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
+  if (pRow->numOfRows == 0) {
+    pRow->numOfRows = 1;
+  }
 }
 
 // todo extract method with copytoSSDataBlock

From 81d3a2c3856ee566a1a7ed6a53ecdfe53ca08b67 Mon Sep 17 00:00:00 2001
From: shenglian zhou <slzhou@taosdata.com>
Date: Sat, 6 Aug 2022 21:39:53 +0800
Subject: [PATCH 10/12] fix: last/first must return not null values

---
 include/libs/function/functionMgt.h                 |  1 +
 source/libs/executor/src/executorimpl.c             | 13 +++++++++----
 source/libs/function/src/functionMgt.c              | 12 ++++++++++++
 tests/script/tsim/parser/function.sim               |  6 +++---
 tests/system-test/2-query/avg.py                    |  2 +-
 .../2-query/distribute_agg_apercentile.py           |  2 +-
 tests/system-test/2-query/distribute_agg_max.py     |  2 +-
 tests/system-test/2-query/distribute_agg_min.py     |  2 +-
 tests/system-test/2-query/distribute_agg_spread.py  |  2 +-
 tests/system-test/2-query/sample.py                 |  4 ++--
 10 files changed, 32 insertions(+), 14 deletions(-)

diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h
index f6f2ceb957..e2846c5974 100644
--- a/include/libs/function/functionMgt.h
+++ b/include/libs/function/functionMgt.h
@@ -202,6 +202,7 @@ bool fmIsForbidStreamFunc(int32_t funcId);
 bool fmIsIntervalInterpoFunc(int32_t funcId);
 bool fmIsInterpFunc(int32_t funcId);
 bool fmIsLastRowFunc(int32_t funcId);
+bool fmIsReturnNotNullFunc(int32_t funcId);
 bool fmIsSelectValueFunc(int32_t funcId);
 bool fmIsSystemInfoFunc(int32_t funcId);
 bool fmIsImplicitTsFunc(int32_t funcId);
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index b99efce106..09094b1b14 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -1437,7 +1437,8 @@ static void setExecutionContext(SOperatorInfo* pOperator, int32_t numOfOutput, u
   pAggInfo->groupId = groupId;
 }
 
-static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_t* rowCellOffset) {
+static void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t numOfExprs, const int32_t* rowCellOffset) {
+  bool returnNotNull = false;
   for (int32_t j = 0; j < numOfExprs; ++j) {
     struct SResultRowEntryInfo* pResInfo = getResultEntryInfo(pRow, j, rowCellOffset);
     if (!isRowEntryInitialized(pResInfo)) {
@@ -1447,9 +1448,13 @@ static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_
     if (pRow->numOfRows < pResInfo->numOfRes) {
       pRow->numOfRows = pResInfo->numOfRes;
     }
+
+    if (fmIsReturnNotNullFunc(pCtx[j].functionId)) {
+      returnNotNull = true;
+    }
   }
   // TODO: if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
-  if (pRow->numOfRows == 0) {
+  if (pRow->numOfRows == 0 && !returnNotNull) {
     pRow->numOfRows = 1;
   }
 }
@@ -1462,7 +1467,7 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
   SFilePage*  page = getBufPage(pBuf, resultRowPosition->pageId);
   SResultRow* pRow = (SResultRow*)((char*)page + resultRowPosition->offset);
 
-  doUpdateNumOfRows(pRow, numOfExprs, rowCellOffset);
+  doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowCellOffset);
   if (pRow->numOfRows == 0) {
     releaseBufPage(pBuf, page);
     return 0;
@@ -1518,7 +1523,7 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprI
 
     SResultRow* pRow = (SResultRow*)((char*)page + pPos->pos.offset);
 
-    doUpdateNumOfRows(pRow, numOfExprs, rowCellOffset);
+    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowCellOffset);
     if (pRow->numOfRows == 0) {
       pGroupResInfo->index += 1;
       releaseBufPage(pBuf, page);
diff --git a/source/libs/function/src/functionMgt.c b/source/libs/function/src/functionMgt.c
index 339ff3808b..88efa7c2ff 100644
--- a/source/libs/function/src/functionMgt.c
+++ b/source/libs/function/src/functionMgt.c
@@ -221,6 +221,18 @@ bool fmIsLastRowFunc(int32_t funcId) {
   return FUNCTION_TYPE_LAST_ROW == funcMgtBuiltins[funcId].type;
 }
 
+bool fmIsReturnNotNullFunc(int32_t funcId) {
+  if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
+    return false;
+  }
+  return FUNCTION_TYPE_LAST == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_LAST_PARTIAL == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_LAST_MERGE == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST_PARTIAL == funcMgtBuiltins[funcId].type ||
+         FUNCTION_TYPE_FIRST_MERGE == funcMgtBuiltins[funcId].type;
+}
+
 bool fmIsSelectValueFunc(int32_t funcId) {
   if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
     return false;
diff --git a/tests/script/tsim/parser/function.sim b/tests/script/tsim/parser/function.sim
index 110901a6e1..9423f53e3a 100644
--- a/tests/script/tsim/parser/function.sim
+++ b/tests/script/tsim/parser/function.sim
@@ -327,8 +327,8 @@ endi
 
 print =================>td-2610
 sql select stddev(k) from tm2 where ts='2020-12-29 18:46:19.109'
-if $rows != 0 then
-  print expect 0, actual:$rows
+if $rows != 1 then
+  print expect 1, actual:$rows
   return -1
 endi
 sql select twa(k) from tm2 where ts='2020-12-29 18:46:19.109'
@@ -406,7 +406,7 @@ if $data00 != 1.378704626 then
 endi
 
 sql select stddev(c) from m1
-if $rows != 0 then
+if $rows != 1 then
   return -1
 endi
 
diff --git a/tests/system-test/2-query/avg.py b/tests/system-test/2-query/avg.py
index ea7c3329ea..2afcc29ac8 100644
--- a/tests/system-test/2-query/avg.py
+++ b/tests/system-test/2-query/avg.py
@@ -295,7 +295,7 @@ class TDTestCase:
         tdSql.checkData(0, 0, 4.500000000)
 
         tdSql.query(f" select avg(c1) from {dbname}.stb1 where c1 is null ")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
 
     def avg_func_filter(self, dbname="db"):
diff --git a/tests/system-test/2-query/distribute_agg_apercentile.py b/tests/system-test/2-query/distribute_agg_apercentile.py
index ad754ad805..34164acfa4 100644
--- a/tests/system-test/2-query/distribute_agg_apercentile.py
+++ b/tests/system-test/2-query/distribute_agg_apercentile.py
@@ -86,7 +86,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select apercentile(c1 , 20) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select apercentile(c1 , 20) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,2.800000000)
diff --git a/tests/system-test/2-query/distribute_agg_max.py b/tests/system-test/2-query/distribute_agg_max.py
index a7b31a2084..f67088d9e6 100644
--- a/tests/system-test/2-query/distribute_agg_max.py
+++ b/tests/system-test/2-query/distribute_agg_max.py
@@ -168,7 +168,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select max(c1) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select max(c1) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,10)
diff --git a/tests/system-test/2-query/distribute_agg_min.py b/tests/system-test/2-query/distribute_agg_min.py
index cc50092451..29e7c5f519 100644
--- a/tests/system-test/2-query/distribute_agg_min.py
+++ b/tests/system-test/2-query/distribute_agg_min.py
@@ -167,7 +167,7 @@ class TDTestCase:
     def distribute_agg_query(self, dbname="testdb"):
         # basic filter
         tdSql.query(f"select min(c1) from {dbname}.stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(f"select min(c1) from {dbname}.stb1 where t1=1")
         tdSql.checkData(0,0,2)
diff --git a/tests/system-test/2-query/distribute_agg_spread.py b/tests/system-test/2-query/distribute_agg_spread.py
index 842a74628d..135a61d194 100644
--- a/tests/system-test/2-query/distribute_agg_spread.py
+++ b/tests/system-test/2-query/distribute_agg_spread.py
@@ -195,7 +195,7 @@ class TDTestCase:
     def distribute_agg_query(self):
         # basic filter
         tdSql.query("select spread(c1) from stb1 where c1 is null")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query("select spread(c1) from stb1 where t1=1")
         tdSql.checkData(0,0,8.000000000)
diff --git a/tests/system-test/2-query/sample.py b/tests/system-test/2-query/sample.py
index b6bdd8926e..f09265c300 100644
--- a/tests/system-test/2-query/sample.py
+++ b/tests/system-test/2-query/sample.py
@@ -743,7 +743,7 @@ class TDTestCase:
         # filter data
 
         tdSql.query(" select sample(c1, 20 ) from t1 where c1 is null ")
-        tdSql.checkRows(0)
+        tdSql.checkRows(1)
 
         tdSql.query(" select sample(c1, 20 ) from t1 where c1 =6 ")
         tdSql.checkRows(1)
@@ -891,4 +891,4 @@ class TDTestCase:
         tdLog.success("%s successfully executed" % __file__)
 
 tdCases.addWindows(__file__, TDTestCase())
-tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
+tdCases.addLinux(__file__, TDTestCase())

From e2aa735572119690c26c53d92f72b204d8641750 Mon Sep 17 00:00:00 2001
From: Xuefeng Tan <1172915550@qq.com>
Date: Sat, 6 Aug 2022 21:56:47 +0800
Subject: [PATCH 11/12] test(go): fix golang example (#15812)

---
 docs/examples/go/connect/afconn/main.go       |  3 +-
 docs/examples/go/connect/cgoexample/main.go   |  3 +-
 docs/examples/go/connect/restexample/main.go  |  3 +-
 docs/examples/go/connect/wrapper/main.go      | 17 ----------
 docs/examples/go/insert/json/main.go          |  6 ++--
 docs/examples/go/insert/line/main.go          |  3 +-
 docs/examples/go/insert/sql/main.go           | 22 ++++++-------
 docs/examples/go/insert/stmt/main.go          |  2 +-
 docs/examples/go/insert/telnet/main.go        |  6 ++--
 docs/examples/go/query/sync/main.go           | 12 +++----
 .../zh/07-develop/03-insert-data/_go_stmt.mdx |  2 +-
 docs/zh/14-reference/11-docker/index.md       |  2 +-
 tests/docs-examples-test/go.sh                | 31 +++++++++++++++++++
 13 files changed, 63 insertions(+), 49 deletions(-)
 delete mode 100644 docs/examples/go/connect/wrapper/main.go
 create mode 100644 tests/docs-examples-test/go.sh

diff --git a/docs/examples/go/connect/afconn/main.go b/docs/examples/go/connect/afconn/main.go
index 3c16212a45..bb2574a01b 100644
--- a/docs/examples/go/connect/afconn/main.go
+++ b/docs/examples/go/connect/afconn/main.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"log"
 
 	"github.com/taosdata/driver-go/v3/af"
 )
@@ -10,7 +11,7 @@ func main() {
 	conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
 	defer conn.Close()
 	if err != nil {
-		fmt.Println("failed to connect, err:", err)
+		log.Fatalln("failed to connect, err:", err)
 	} else {
 		fmt.Println("connected")
 	}
diff --git a/docs/examples/go/connect/cgoexample/main.go b/docs/examples/go/connect/cgoexample/main.go
index 57b8d5d91f..881cf15ee3 100644
--- a/docs/examples/go/connect/cgoexample/main.go
+++ b/docs/examples/go/connect/cgoexample/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"database/sql"
 	"fmt"
+	"log"
 
 	_ "github.com/taosdata/driver-go/v3/taosSql"
 )
@@ -11,7 +12,7 @@ func main() {
 	var taosDSN = "root:taosdata@tcp(localhost:6030)/"
 	taos, err := sql.Open("taosSql", taosDSN)
 	if err != nil {
-		fmt.Println("failed to connect TDengine, err:", err)
+		log.Fatalln("failed to connect TDengine, err:", err)
 		return
 	}
 	fmt.Println("Connected")
diff --git a/docs/examples/go/connect/restexample/main.go b/docs/examples/go/connect/restexample/main.go
index f10151f04f..67a129bf9c 100644
--- a/docs/examples/go/connect/restexample/main.go
+++ b/docs/examples/go/connect/restexample/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"database/sql"
 	"fmt"
+	"log"
 
 	_ "github.com/taosdata/driver-go/v3/taosRestful"
 )
@@ -11,7 +12,7 @@ func main() {
 	var taosDSN = "root:taosdata@http(localhost:6041)/"
 	taos, err := sql.Open("taosRestful", taosDSN)
 	if err != nil {
-		fmt.Println("failed to connect TDengine, err:", err)
+		log.Fatalln("failed to connect TDengine, err:", err)
 		return
 	}
 	fmt.Println("Connected")
diff --git a/docs/examples/go/connect/wrapper/main.go b/docs/examples/go/connect/wrapper/main.go
deleted file mode 100644
index a459539db3..0000000000
--- a/docs/examples/go/connect/wrapper/main.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package main
-
-import (
-	"fmt"
-
-	"github.com/taosdata/driver-go/v3/wrapper"
-)
-
-func main() {
-	conn, err := wrapper.TaosConnect("localhost", "root", "taosdata", "", 6030)
-	defer wrapper.TaosClose(conn)
-	if err != nil {
-		fmt.Println("fail to connect, err:", err)
-	} else {
-		fmt.Println("connected")
-	}
-}
diff --git a/docs/examples/go/insert/json/main.go b/docs/examples/go/insert/json/main.go
index 805c123342..e470cc6bd4 100644
--- a/docs/examples/go/insert/json/main.go
+++ b/docs/examples/go/insert/json/main.go
@@ -1,7 +1,7 @@
 package main
 
 import (
-	"fmt"
+	"log"
 
 	"github.com/taosdata/driver-go/v3/af"
 )
@@ -20,7 +20,7 @@ func prepareDatabase(conn *af.Connector) {
 func main() {
 	conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
 	if err != nil {
-		fmt.Println("fail to connect, err:", err)
+		log.Fatalln("fail to connect, err:", err)
 	}
 	defer conn.Close()
 	prepareDatabase(conn)
@@ -32,6 +32,6 @@ func main() {
 
 	err = conn.OpenTSDBInsertJsonPayload(payload)
 	if err != nil {
-		fmt.Println("insert error:", err)
+		log.Fatalln("insert error:", err)
 	}
 }
diff --git a/docs/examples/go/insert/line/main.go b/docs/examples/go/insert/line/main.go
index 1a09edc40c..a270ab2633 100644
--- a/docs/examples/go/insert/line/main.go
+++ b/docs/examples/go/insert/line/main.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"log"
 
 	"github.com/taosdata/driver-go/v3/af"
 )
@@ -33,6 +34,6 @@ func main() {
 
 	err = conn.InfluxDBInsertLines(lines, "ms")
 	if err != nil {
-		fmt.Println("insert error:", err)
+		log.Fatalln("insert error:", err)
 	}
 }
diff --git a/docs/examples/go/insert/sql/main.go b/docs/examples/go/insert/sql/main.go
index a493ef5fb0..57f881314c 100644
--- a/docs/examples/go/insert/sql/main.go
+++ b/docs/examples/go/insert/sql/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"database/sql"
 	"fmt"
+	"log"
 
 	_ "github.com/taosdata/driver-go/v3/taosRestful"
 )
@@ -10,28 +11,26 @@ import (
 func createStable(taos *sql.DB) {
 	_, err := taos.Exec("CREATE DATABASE power")
 	if err != nil {
-		fmt.Println("failed to create database, err:", err)
+		log.Fatalln("failed to create database, err:", err)
 	}
 	_, err = taos.Exec("CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (location BINARY(64), groupId INT)")
 	if err != nil {
-		fmt.Println("failed to create stable, err:", err)
+		log.Fatalln("failed to create stable, err:", err)
 	}
 }
 
 func insertData(taos *sql.DB) {
-	sql := `INSERT INTO power.d1001 USING power.meters TAGS(California.SanFrancisco, 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000)
-	power.d1002 USING power.meters TAGS(California.SanFrancisco, 3) VALUES ('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000)
-	power.d1003 USING power.meters TAGS(California.LosAngeles, 2) VALUES ('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000) ('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000)
-	power.d1004 USING power.meters TAGS(California.LosAngeles, 3) VALUES ('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000) ('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)`
+	sql := `INSERT INTO power.d1001 USING power.meters TAGS('California.SanFrancisco', 2) VALUES ('2018-10-03 14:38:05.000', 10.30000, 219, 0.31000) ('2018-10-03 14:38:15.000', 12.60000, 218, 0.33000) ('2018-10-03 14:38:16.800', 12.30000, 221, 0.31000)
+	power.d1002 USING power.meters TAGS('California.SanFrancisco', 3) VALUES ('2018-10-03 14:38:16.650', 10.30000, 218, 0.25000)
+	power.d1003 USING power.meters TAGS('California.LosAngeles', 2) VALUES ('2018-10-03 14:38:05.500', 11.80000, 221, 0.28000) ('2018-10-03 14:38:16.600', 13.40000, 223, 0.29000)
+	power.d1004 USING power.meters TAGS('California.LosAngeles', 3) VALUES ('2018-10-03 14:38:05.000', 10.80000, 223, 0.29000) ('2018-10-03 14:38:06.500', 11.50000, 221, 0.35000)`
 	result, err := taos.Exec(sql)
 	if err != nil {
-		fmt.Println("failed to insert, err:", err)
-		return
+		log.Fatalln("failed to insert, err:", err)
 	}
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		fmt.Println("failed to get affected rows, err:", err)
-		return
+		log.Fatalln("failed to get affected rows, err:", err)
 	}
 	fmt.Println("RowsAffected", rowsAffected)
 }
@@ -40,8 +39,7 @@ func main() {
 	var taosDSN = "root:taosdata@http(localhost:6041)/"
 	taos, err := sql.Open("taosRestful", taosDSN)
 	if err != nil {
-		fmt.Println("failed to connect TDengine, err:", err)
-		return
+		log.Fatalln("failed to connect TDengine, err:", err)
 	}
 	defer taos.Close()
 	createStable(taos)
diff --git a/docs/examples/go/insert/stmt/main.go b/docs/examples/go/insert/stmt/main.go
index d23a6e345c..bcbb8e632b 100644
--- a/docs/examples/go/insert/stmt/main.go
+++ b/docs/examples/go/insert/stmt/main.go
@@ -5,8 +5,8 @@ import (
 	"time"
 
 	"github.com/taosdata/driver-go/v3/af"
-	"github.com/taosdata/driver-go/v3/af/param"
 	"github.com/taosdata/driver-go/v3/common"
+	"github.com/taosdata/driver-go/v3/common/param"
 )
 
 func checkErr(err error, prompt string) {
diff --git a/docs/examples/go/insert/telnet/main.go b/docs/examples/go/insert/telnet/main.go
index 4cf4375db5..63e9a56dbf 100644
--- a/docs/examples/go/insert/telnet/main.go
+++ b/docs/examples/go/insert/telnet/main.go
@@ -1,7 +1,7 @@
 package main
 
 import (
-	"fmt"
+	"log"
 
 	"github.com/taosdata/driver-go/v3/af"
 )
@@ -20,7 +20,7 @@ func prepareDatabase(conn *af.Connector) {
 func main() {
 	conn, err := af.Open("localhost", "root", "taosdata", "", 6030)
 	if err != nil {
-		fmt.Println("fail to connect, err:", err)
+		log.Fatalln("fail to connect, err:", err)
 	}
 	defer conn.Close()
 	prepareDatabase(conn)
@@ -37,6 +37,6 @@ func main() {
 
 	err = conn.OpenTSDBInsertTelnetLines(lines)
 	if err != nil {
-		fmt.Println("insert error:", err)
+		log.Fatalln("insert error:", err)
 	}
 }
diff --git a/docs/examples/go/query/sync/main.go b/docs/examples/go/query/sync/main.go
index add422e385..e37164f47f 100644
--- a/docs/examples/go/query/sync/main.go
+++ b/docs/examples/go/query/sync/main.go
@@ -2,7 +2,7 @@ package main
 
 import (
 	"database/sql"
-	"fmt"
+	"log"
 	"time"
 
 	_ "github.com/taosdata/driver-go/v3/taosRestful"
@@ -12,14 +12,12 @@ func main() {
 	var taosDSN = "root:taosdata@http(localhost:6041)/power"
 	taos, err := sql.Open("taosRestful", taosDSN)
 	if err != nil {
-		fmt.Println("failed to connect TDengine, err:", err)
-		return
+		log.Fatalln("failed to connect TDengine, err:", err)
 	}
 	defer taos.Close()
 	rows, err := taos.Query("SELECT ts, current FROM meters LIMIT 2")
 	if err != nil {
-		fmt.Println("failed to select from table, err:", err)
-		return
+		log.Fatalln("failed to select from table, err:", err)
 	}
 
 	defer rows.Close()
@@ -30,9 +28,9 @@ func main() {
 		}
 		err := rows.Scan(&r.ts, &r.current)
 		if err != nil {
-			fmt.Println("scan error:\n", err)
+			log.Fatalln("scan error:\n", err)
 			return
 		}
-		fmt.Println(r.ts, r.current)
+		log.Fatalln(r.ts, r.current)
 	}
 }
diff --git a/docs/zh/07-develop/03-insert-data/_go_stmt.mdx b/docs/zh/07-develop/03-insert-data/_go_stmt.mdx
index d65a96549f..db4e227c43 100644
--- a/docs/zh/07-develop/03-insert-data/_go_stmt.mdx
+++ b/docs/zh/07-develop/03-insert-data/_go_stmt.mdx
@@ -3,6 +3,6 @@
 ```
 
 :::tip
-driver-go 的模块 `github.com/taosdata/driver-go/v2/wrapper` 是 C 接口的底层封装。使用这个模块也可以实现参数绑定写入。
+driver-go 的模块 `github.com/taosdata/driver-go/v3/wrapper` 是 C 接口的底层封装。使用这个模块也可以实现参数绑定写入。
 
 :::
diff --git a/docs/zh/14-reference/11-docker/index.md b/docs/zh/14-reference/11-docker/index.md
index d0b7f14204..e40f0dbd35 100644
--- a/docs/zh/14-reference/11-docker/index.md
+++ b/docs/zh/14-reference/11-docker/index.md
@@ -147,7 +147,7 @@ import (
     "fmt"
     "time"
 
-    _ "github.com/taosdata/driver-go/v2/taosSql"
+    _ "github.com/taosdata/driver-go/v3/taosSql"
 )
 
 type config struct {
diff --git a/tests/docs-examples-test/go.sh b/tests/docs-examples-test/go.sh
new file mode 100644
index 0000000000..185661e8a7
--- /dev/null
+++ b/tests/docs-examples-test/go.sh
@@ -0,0 +1,31 @@
+#!/bin/bash
+
+set -e
+
+taosd >>/dev/null 2>&1 &
+taosadapter >>/dev/null 2>&1 &
+
+cd ../../docs/examples/go
+
+go mod tidy
+
+go run ./connect/afconn/main.go
+go run ./connect/cgoexample/main.go
+go run ./connect/restexample/main.go
+
+taos -s "drop database if exists test"
+go run ./insert/json/main.go
+taos -s "drop database if exists test"
+go run ./insert/line/main.go
+taos -s "drop database if exists power"
+go run ./insert/sql/main.go
+taos -s "drop database if exists power"
+go run ./insert/stmt/main.go
+taos -s "drop database if exists test"
+go run ./insert/telnet/main.go
+
+go run ./query/sync/main.go
+
+taos -s "drop topic if exists example_tmq_topic"
+taos -s "drop database if exists example_tmq"
+go run ./sub/main.go

From be8c3628c5bc126d2c82dec67f453f5329138915 Mon Sep 17 00:00:00 2001
From: shenglian zhou <slzhou@taosdata.com>
Date: Sun, 7 Aug 2022 06:08:12 +0800
Subject: [PATCH 12/12] fix: trigger ci test

---
 source/libs/executor/src/executorimpl.c | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 5666dded83..cffd38c98a 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -1453,7 +1453,8 @@ static void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t nu
       returnNotNull = true;
     }
   }
-  // TODO: if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
+  // if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
+  //  except for first/last, which require not null output, output no rows
   if (pRow->numOfRows == 0 && !returnNotNull) {
     pRow->numOfRows = 1;
   }