Merge remote-tracking branch 'origin/3.0' into enh/TD-19660

include/common/tdataformat.h

@@ -234,8 +234,6 @@ struct STag {
 // Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap.
 #define TD_SUPPORT_BITMAP
 
-#define TASSERT(x) ASSERT(x)
-
 #define STR_TO_VARSTR(x, str)                     \
   do {                                            \
     VarDataLenT __len = (VarDataLenT)strlen(str); \

include/common/trow.h

@@ -255,7 +255,7 @@ static FORCE_INLINE void *tdGetBitmapAddrKv(STSRow *pRow, col_id_t nKvCols) {
 void   *tdGetBitmapAddr(STSRow *pRow, uint8_t rowType, uint32_t flen, col_id_t nKvCols);
 int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TDRowValT valType, int8_t bitmapMode);
 int32_t tdSetBitmapValTypeII(void *pBitmap, int16_t colIdx, TDRowValT valType);
-bool    tdIsBitmapValTypeNorm(const void *pBitmap, int16_t idx, int8_t bitmapMode);
+// bool    tdIsBitmapValTypeNorm(const void *pBitmap, int16_t idx, int8_t bitmapMode);
 int32_t tdGetBitmapValType(const void *pBitmap, int16_t colIdx, TDRowValT *pValType, int8_t bitmapMode);
 
 // ----------------- Tuple row structure(STpRow)

include/libs/stream/streamState.h

@@ -68,9 +68,10 @@ int32_t streamStateSessionClear(SStreamState* pState);
 int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, const void** pVal, int32_t* pVLen);
 int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
                                       state_key_cmpr_fn fn, void** pVal, int32_t* pVLen);
+int32_t streamStateSessionGetKey(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey);
 
 SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key);
-SStreamStateCur* streamStateSessionSeekKeyPrev(SStreamState* pState, const SSessionKey* key);
+SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, const SSessionKey* key);
 SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKey* key);
 
 int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);

source/common/src/tdatablock.c

@@ -1858,7 +1858,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
             char*   pData = colDataGetVarData(pColInfoData, j);
             int32_t dataSize = TMIN(sizeof(pBuf), varDataLen(pData));
             memset(pBuf, 0, dataSize);
-            taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
+            (void)taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
             printf(" %15s |", pBuf);
           } break;
           default:
@@ -1946,7 +1946,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
           char*   pData = colDataGetVarData(pColInfoData, j);
           int32_t dataSize = TMIN(sizeof(pBuf), varDataLen(pData));
           memset(pBuf, 0, sizeof(pBuf));
-          taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
+          (void)taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
           len += snprintf(dumpBuf + len, size - len, " %15s |", pBuf);
           if (len >= size - 1) return dumpBuf;
         } break;
@@ -2053,7 +2053,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
           case TSDB_DATA_TYPE_JSON:
           case TSDB_DATA_TYPE_MEDIUMBLOB:
             uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
-            TASSERT(0);
+            ASSERT(0);
             break;
           default:
             if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
@@ -2084,7 +2084,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
               }
             } else {
               uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
-              TASSERT(0);
+              ASSERT(0);
             }
             break;
         }

source/common/src/trow.c

@@ -481,7 +481,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
 
 int32_t tdGetBitmapValTypeII(const void *pBitmap, int16_t colIdx, TDRowValT *pValType) {
   if (!pBitmap || colIdx < 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -503,7 +503,7 @@ int32_t tdGetBitmapValTypeII(const void *pBitmap, int16_t colIdx, TDRowValT *pVa
       *pValType = ((*pDestByte) & 0x03);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -512,7 +512,7 @@ int32_t tdGetBitmapValTypeII(const void *pBitmap, int16_t colIdx, TDRowValT *pVa
 
 int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pValType) {
   if (!pBitmap || colIdx < 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -546,7 +546,7 @@ int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pVal
       *pValType = ((*pDestByte) & 0x01);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -555,7 +555,7 @@ int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pVal
 
 int32_t tdSetBitmapValTypeI(void *pBitmap, int16_t colIdx, TDRowValT valType) {
   if (!pBitmap || colIdx < 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -598,7 +598,7 @@ int32_t tdSetBitmapValTypeI(void *pBitmap, int16_t colIdx, TDRowValT valType) {
       // *pDestByte |= (valType);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -607,7 +607,7 @@ int32_t tdSetBitmapValTypeI(void *pBitmap, int16_t colIdx, TDRowValT valType) {
 
 int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int32_t offset, int16_t colIdx) {
 #ifdef TD_SUPPORT_BITMAP
-  TASSERT(colIdx < tdRowGetNCols(pRow) - 1);
+  ASSERT(colIdx < tdRowGetNCols(pRow) - 1);
   if (tdGetBitmapValType(pBitmap, colIdx, &output->valType, 0) != TSDB_CODE_SUCCESS) {
     output->valType = TD_VTYPE_NONE;
     return terrno;
@@ -621,7 +621,7 @@ int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int32_
     output->val = POINTER_SHIFT(pRow, offset);
   }
 #else
-  TASSERT(0);
+  ASSERT(0);
   if (offset < 0) {
     terrno = TSDB_CODE_INVALID_PARA;
     output->valType = TD_VTYPE_NONE;
@@ -671,7 +671,7 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
       return terrno;
     }
 #else
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
 #endif
@@ -709,7 +709,7 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
 int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val, bool isCopyVarData,
                               int8_t colType, int16_t colIdx, int32_t offset, col_id_t colId) {
   if ((offset < (int32_t)sizeof(SKvRowIdx)) || (colIdx < 1)) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -798,7 +798,7 @@ int32_t tdSRowSetExtendedInfo(SRowBuilder *pBuilder, int32_t nCols, int32_t nBou
   pBuilder->nCols = nCols;
   pBuilder->nBoundCols = nBoundCols;
   if (pBuilder->flen <= 0 || pBuilder->nCols <= 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -820,7 +820,7 @@ int32_t tdSRowSetExtendedInfo(SRowBuilder *pBuilder, int32_t nCols, int32_t nBou
 int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
   pBuilder->pBuf = (STSRow *)pBuf;
   if (!pBuilder->pBuf) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -831,7 +831,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
   TD_ROW_SET_INFO(pBuilder->pBuf, 0);
   TD_ROW_SET_TYPE(pBuilder->pBuf, pBuilder->rowType);
 
-  TASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
+  ASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
 
   uint32_t len = 0;
   switch (pBuilder->rowType) {
@@ -857,7 +857,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
       TD_ROW_SET_NCOLS(pBuilder->pBuf, pBuilder->nBoundCols);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -868,12 +868,12 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
 int32_t tdSRowGetBuf(SRowBuilder *pBuilder, void *pBuf) {
   pBuilder->pBuf = (STSRow *)pBuf;
   if (!pBuilder->pBuf) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
 
-  TASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
+  ASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
 
   uint32_t len = 0;
   switch (pBuilder->rowType) {
@@ -888,7 +888,7 @@ int32_t tdSRowGetBuf(SRowBuilder *pBuilder, void *pBuf) {
 #endif
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -908,7 +908,7 @@ int32_t tdSRowSetTpInfo(SRowBuilder *pBuilder, int32_t nCols, int32_t flen) {
   pBuilder->flen = flen;
   pBuilder->nCols = nCols;
   if (pBuilder->flen <= 0 || pBuilder->nCols <= 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -927,7 +927,7 @@ int32_t tdSRowSetInfo(SRowBuilder *pBuilder, int32_t nCols, int32_t nBoundCols,
   pBuilder->nCols = nCols;
   pBuilder->nBoundCols = nBoundCols;
   if (pBuilder->flen <= 0 || pBuilder->nCols <= 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -956,13 +956,13 @@ int32_t tdGetBitmapValType(const void *pBitmap, int16_t colIdx, TDRowValT *pValT
       tdGetBitmapValTypeI(pBitmap, colIdx, pValType);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return TSDB_CODE_FAILED;
   }
   return TSDB_CODE_SUCCESS;
 }
-
+#if 0
 bool tdIsBitmapValTypeNorm(const void *pBitmap, int16_t idx, int8_t bitmapMode) {
   TDRowValT valType = 0;
   tdGetBitmapValType(pBitmap, idx, &valType, bitmapMode);
@@ -971,10 +971,11 @@ bool tdIsBitmapValTypeNorm(const void *pBitmap, int16_t idx, int8_t bitmapMode)
   }
   return false;
 }
+#endif
 
 int32_t tdSetBitmapValTypeII(void *pBitmap, int16_t colIdx, TDRowValT valType) {
   if (!pBitmap || colIdx < 0) {
-    TASSERT(0);
+    ASSERT(0);
     terrno = TSDB_CODE_INVALID_PARA;
     return terrno;
   }
@@ -1001,7 +1002,7 @@ int32_t tdSetBitmapValTypeII(void *pBitmap, int16_t colIdx, TDRowValT valType) {
       // *pDestByte |= (valType);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return terrno;
   }
@@ -1018,7 +1019,7 @@ int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TDRowValT valType, int
       tdSetBitmapValTypeI(pBitmap, colIdx, valType);
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       terrno = TSDB_CODE_INVALID_PARA;
       return TSDB_CODE_FAILED;
   }

source/dnode/mnode/impl/src/mndDb.c

@@ -113,7 +113,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
   SDB_SET_INT8(pRaw, dataPos, pDb->cfg.hashMethod, _OVER)
   SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfRetensions, _OVER)
   for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
-    TASSERT(taosArrayGetSize(pDb->cfg.pRetensions) == pDb->cfg.numOfRetensions);
+    ASSERT(taosArrayGetSize(pDb->cfg.pRetensions) == pDb->cfg.numOfRetensions);
     SRetention *pRetension = taosArrayGet(pDb->cfg.pRetensions, i);
     SDB_SET_INT64(pRaw, dataPos, pRetension->freq, _OVER)
     SDB_SET_INT64(pRaw, dataPos, pRetension->keep, _OVER)

source/libs/executor/src/timewindowoperator.c

@@ -3594,9 +3594,7 @@ void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endT
   pKey->win.skey = startTs;
   pKey->win.ekey = endTs;
   pKey->groupId = groupId;
-  SStreamStateCur* pCur = streamStateSessionGetCur(pAggSup->pState, pKey);
-  int32_t          code = streamStateSessionGetKVByCur(pCur, pKey, NULL, 0);
-  streamStateFreeCur(pCur);
+  int32_t code = streamStateSessionGetKey(pAggSup->pState, pKey, pKey);
   if (code != TSDB_CODE_SUCCESS) {
     SET_SESSION_WIN_KEY_INVALID(pKey);
   }

source/libs/function/inc/tudfInt.h

@@ -35,7 +35,7 @@ enum {
 };
 
 typedef struct SUdfSetupRequest {
-  char udfName[TSDB_FUNC_NAME_LEN];
+  char udfName[TSDB_FUNC_NAME_LEN + 1];
 } SUdfSetupRequest;
 
 typedef struct SUdfSetupResponse {

source/libs/function/src/tudf.c

@@ -315,7 +315,7 @@ enum { UV_TASK_CONNECT = 0, UV_TASK_REQ_RSP = 1, UV_TASK_DISCONNECT = 2 };
 
 int64_t gUdfTaskSeqNum = 0;
 typedef struct SUdfcFuncStub {
-  char           udfName[TSDB_FUNC_NAME_LEN];
+  char           udfName[TSDB_FUNC_NAME_LEN + 1];
   UdfcFuncHandle handle;
   int32_t        refCount;
   int64_t        lastRefTime;
@@ -353,7 +353,7 @@ typedef struct SUdfcUvSession {
   int32_t outputLen;
   int32_t bufSize;
 
-  char udfName[TSDB_FUNC_NAME_LEN];
+  char udfName[TSDB_FUNC_NAME_LEN + 1];
 } SUdfcUvSession;
 
 typedef struct SClientUvTaskNode {
@@ -898,7 +898,7 @@ int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) {
   int32_t code = 0;
   uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
   SUdfcFuncStub key = {0};
-  strcpy(key.udfName, udfName);
+  strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
   int32_t stubIndex = taosArraySearchIdx(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
   if (stubIndex != -1) {
     SUdfcFuncStub *foundStub = taosArrayGet(gUdfdProxy.udfStubs, stubIndex);
@@ -936,7 +936,7 @@ int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) {
 void releaseUdfFuncHandle(char *udfName) {
   uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
   SUdfcFuncStub key = {0};
-  strcpy(key.udfName, udfName);
+  strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
   SUdfcFuncStub *foundStub = taosArraySearch(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
   if (!foundStub) {
     uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
@@ -1446,6 +1446,7 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
         QUEUE_INSERT_TAIL(connTaskQueue, &uvTask->connTaskQueue);
         int err = uv_write(write, (uv_stream_t *)pipe, &uvTask->reqBuf, 1, onUdfcPipeWrite);
         if (err != 0) {
+          taosMemoryFree(write);
           fnError("udfc event loop start req_rsp task uv_write failed. uvtask: %p, code: %s", uvTask, uv_strerror(err));
         }
         code = err;
@@ -1637,7 +1638,7 @@ int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
   task->session->outputType = rsp->outputType;
   task->session->outputLen = rsp->outputLen;
   task->session->bufSize = rsp->bufSize;
-  strcpy(task->session->udfName, udfName);
+  strncpy(task->session->udfName, udfName, TSDB_FUNC_NAME_LEN);
   if (task->errCode != 0) {
     fnError("failed to setup udf. udfname: %s, err: %d", udfName, task->errCode)
   } else {

source/libs/function/src/udfd.c

@@ -71,7 +71,7 @@ typedef struct SUdf {
   uv_cond_t  condReady;
   bool resident;
 
-  char    name[TSDB_FUNC_NAME_LEN];
+  char    name[TSDB_FUNC_NAME_LEN + 1];
   int8_t  funcType;
   int8_t  scriptType;
   int8_t  outputType;
@@ -188,11 +188,12 @@ void udfdProcessSetupRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
     SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
     udfNew->refCount = 1;
     udfNew->state = UDF_STATE_INIT;
-
     uv_mutex_init(&udfNew->lock);
     uv_cond_init(&udfNew->condReady);
+
     udf = udfNew;
-    taosHashPut(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName), &udfNew, sizeof(&udfNew));
+    SUdf** pUdf = &udf;
+    taosHashPut(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName), pUdf, POINTER_BYTES);
     uv_mutex_unlock(&global.udfsMutex);
   }
 
@@ -246,7 +247,7 @@ void udfdProcessSetupRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
 
 void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
   SUdfCallRequest *call = &request->call;
-  fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType, call->udfHandle);
+  fnDebug("call request. call type %d, handle: %" PRIx64 ", seq num %" PRId64 , call->callType, call->udfHandle, request->seqNum);
   SUdfcFuncHandle * handle = (SUdfcFuncHandle *)(call->udfHandle);
   SUdf *            udf = handle->udf;
   SUdfResponse      response = {0};
@@ -372,7 +373,7 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
   }
   taosMemoryFree(handle);
 
-  SUdfResponse  response;
+  SUdfResponse  response = {0};
   SUdfResponse *rsp = &response;
   rsp->seqNum = request->seqNum;
   rsp->type = request->type;
@@ -428,7 +429,9 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
   } else if (msgInfo->rpcType == UDFD_RPC_RETRIVE_FUNC) {
     SRetrieveFuncRsp retrieveRsp = {0};
     tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp);
-
+    if (retrieveRsp.pFuncInfos == NULL) {
+      goto _return;
+    }
     SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
     SUdf *     udf = msgInfo->param;
     udf->funcType = pFuncInfo->funcType;
@@ -540,7 +543,7 @@ int32_t udfdConnectToMnode() {
 }
 
 int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
-  strcpy(udf->name, udfName);
+  strncpy(udf->name, udfName, TSDB_FUNC_NAME_LEN);
   int32_t err = 0;
 
   err = udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf);
@@ -880,6 +883,8 @@ static int32_t udfdUvInit() {
   uv_loop_t *loop = taosMemoryMalloc(sizeof(uv_loop_t));
   if (loop) {
     uv_loop_init(loop);
+  } else {
+    return -1;
   }
   global.loop = loop;
 
@@ -901,12 +906,12 @@ static int32_t udfdUvInit() {
   if ((r = uv_pipe_bind(&global.listeningPipe, global.listenPipeName))) {
     fnError("Bind error %s", uv_err_name(r));
     removeListeningPipe();
-    return -1;
+    return -2;
   }
   if ((r = uv_listen((uv_stream_t *)&global.listeningPipe, 128, udfdOnNewConnection))) {
     fnError("Listen error %s", uv_err_name(r));
     removeListeningPipe();
-    return -2;
+    return -3;
   }
   return 0;
 }
@@ -962,6 +967,7 @@ int32_t udfdInitResidentFuncs() {
   while ((token = strtok_r(pSave, ",", &pSave)) != NULL) {
     char func[TSDB_FUNC_NAME_LEN+1] = {0};
     strncpy(func, token, TSDB_FUNC_NAME_LEN);
+    fnInfo("udfd add resident function %s", func);
     taosArrayPush(global.residentFuncs, func);
   }
 

source/libs/function/test/runUdf.c

@@ -110,8 +110,11 @@ int aggregateFuncTest() {
   taosArrayDestroy(pBlock->pDataBlock);
 
   doCallUdfAggFinalize(handle, &newBuf, &resultBuf);
+  if (resultBuf.buf != NULL) {
     fprintf(stderr, "agg result: %f\n", *(double *)resultBuf.buf);
-
+  } else {
+    fprintf(stderr, "result buffer is null");
+  }
   freeUdfInterBuf(&buf);
   freeUdfInterBuf(&newBuf);
   freeUdfInterBuf(&resultBuf);

source/libs/index/src/indexComm.c

@@ -367,7 +367,7 @@ int32_t idxConvertData(void* src, int8_t type, void** dst) {
       tlen = taosEncodeBinary(dst, src, strlen(src));
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       break;
   }
   *dst = (char*)*dst - tlen;
@@ -459,7 +459,7 @@ int32_t idxConvertDataToStr(void* src, int8_t type, void** dst) {
       *dst = (char*)*dst - tlen;
       break;
     default:
-      TASSERT(0);
+      ASSERT(0);
       break;
   }
   return tlen;

source/libs/stream/src/streamState.c

@@ -526,7 +526,7 @@ int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key) {
   return tdbTbDelete(pState->pSessionStateDb, &sKey, sizeof(SStateSessionKey), &pState->txn);
 }
 
-SStreamStateCur* streamStateSessionSeekKeyPrev(SStreamState* pState, const SSessionKey* key) {
+SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, const SSessionKey* key) {
   SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
   if (pCur == NULL) {
     return NULL;
@@ -544,7 +544,7 @@ SStreamStateCur* streamStateSessionSeekKeyPrev(SStreamState* pState, const SSess
     streamStateFreeCur(pCur);
     return NULL;
   }
-  if (c > 0) return pCur;
+  if (c >= 0) return pCur;
 
   if (tdbTbcMoveToPrev(pCur->pCur) < 0) {
     streamStateFreeCur(pCur);
@@ -572,7 +572,7 @@ SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSess
     streamStateFreeCur(pCur);
     return NULL;
   }
-  if (c > 0) return pCur;
+  if (c < 0) return pCur;
 
   if (tdbTbcMoveToNext(pCur->pCur) < 0) {
     streamStateFreeCur(pCur);
@@ -630,7 +630,7 @@ SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKe
     streamStateCurPrev(pState, pCur);
     SSessionKey tmpKey = *key;
     int32_t     code = streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
-    if (code == TSDB_CODE_SUCCESS && sessionKeyCmpr(key, &tmpKey) == 0) {
+    if (code == 0 && sessionKeyCmpr(key, &tmpKey) == 0) {
       resKey = tmpKey;
     } else {
       break;
@@ -640,9 +640,28 @@ SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKe
   return streamStateSessionGetRanomCur(pState, &resKey);
 }
 
+int32_t streamStateSessionGetKey(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
+  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
+  SSessionKey      resKey = *key;
+  int32_t          res = -1;
+  while (1) {
+    SSessionKey tmpKey = *key;
+    int32_t     code = streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
+    if (code == 0 && sessionKeyCmpr(key, &tmpKey) == 0) {
+      res = 0;
+      resKey = tmpKey;
+    } else {
+      break;
+    }
+    streamStateCurPrev(pState, pCur);
+  }
+  *curKey = resKey;
+  return res;
+}
+
 int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen) {
   // todo refactor
-  SStreamStateCur* pCur = streamStateSessionGetCur(pState, key);
+  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
   int32_t          size = *pVLen;
   void*            tmp = NULL;
   *pVal = tdbRealloc(NULL, size);
@@ -659,7 +678,7 @@ int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key,
 int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
                                       state_key_cmpr_fn fn, void** pVal, int32_t* pVLen) {
   // todo refactor
-  int32_t     res = TSDB_CODE_SUCCESS;
+  int32_t     res = 0;
   SSessionKey tmpKey = *key;
   int32_t     valSize = *pVLen;
   void*       tmp = tdbRealloc(NULL, valSize);
@@ -667,21 +686,14 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
     return -1;
   }
 
-  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
+  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentPrev(pState, key);
   int32_t          code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
-  if (code == TSDB_CODE_SUCCESS) {
+  if (code == 0) {
+    if (key->win.skey <= tmpKey.win.skey && tmpKey.win.ekey <= key->win.ekey) {
       memcpy(tmp, *pVal, valSize);
-    *pVal = tmp;
-    streamStateFreeCur(pCur);
-    return res;
+      goto _end;
     }
-  streamStateFreeCur(pCur);
 
-  streamStateSessionPut(pState, key, NULL, 0);
-  pCur = streamStateSessionGetRanomCur(pState, key);
-  streamStateCurPrev(pState, pCur);
-  code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
-  if (code == TSDB_CODE_SUCCESS) {
     void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
     if (fn(pKeyData, stateKey) == true) {
       memcpy(tmp, *pVal, valSize);
@@ -689,11 +701,9 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
     }
   }
 
-  streamStateFreeCur(pCur);
-  *key = tmpKey;
-  pCur = streamStateSessionSeekKeyNext(pState, key);
+  streamStateCurNext(pState, pCur);
   code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
-  if (code == TSDB_CODE_SUCCESS) {
+  if (code == 0) {
     void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
     if (fn(pKeyData, stateKey) == true) {
       memcpy(tmp, *pVal, valSize);
@@ -708,7 +718,6 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
 _end:
 
   *pVal = tmp;
-  streamStateSessionDel(pState, &tmpKey);
   streamStateFreeCur(pCur);
   return res;
 }

source/util/src/tconfig.c

@@ -944,6 +944,7 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
     if (taosReadFile(pFile, buf, fileSize) <= 0) {
       taosCloseFile(&pFile);
       uError("load json file error: %s", filepath);
+      taosMemoryFreeClear(buf);
       return -1;
     }
     taosCloseFile(&pFile);
@@ -953,6 +954,7 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
       if (jsonParseError != NULL) {
         uError("load json file parse error: %s", jsonParseError);
       }
+      taosMemoryFreeClear(buf);
       return -1;
     }
     taosMemoryFreeClear(buf);

tools/shell/src/shellAuto.c

@@ -562,23 +562,15 @@ void parseCommand(SWords* command, bool pattern) {
 
 // free SShellCmd
 void freeCommand(SWords* command) {
-  SWord* word = command->head;
-  if (word == NULL) {
-    return;
-  }
-
+  SWord* item = command->head;
   // loop
-  while (word->next) {
-    SWord* tmp = word;
-    word = word->next;
+  while (item) {
+    SWord* tmp = item;
+    item = item->next;
     // if malloc need free
     if (tmp->free && tmp->word) taosMemoryFree(tmp->word);
     taosMemoryFree(tmp);
   }
-
-  // if malloc need free
-  if (word->free && word->word) taosMemoryFree(word->word);
-  taosMemoryFree(word);
 }
 
 void GenerateVarType(int type, char** p, int count) {
@@ -1204,11 +1196,11 @@ bool nextMatchCommand(TAOS* con, SShellCmd* cmd, SWords* firstMatch) {
 #endif
 
   // free
+  freeCommand(input);
   if (input->source) {
     taosMemoryFree(input->source);
     input->source = NULL;
   }
-  freeCommand(input);
   taosMemoryFree(input);
 
   return true;
@@ -1377,7 +1369,7 @@ bool appendAfterSelect(TAOS* con, SShellCmd* cmd, char* sql, int32_t len) {
   bool  ret = false;
   if (from == NULL) {
     bool fieldEnd = fieldsInputEnd(p);
-    // cheeck fields input end then insert from keyword
+    // check fields input end then insert from keyword
     if (fieldEnd && p[len - 1] == ' ') {
       shellInsertChar(cmd, "from", 4);
       taosMemoryFree(p);

tools/shell/src/shellCommand.c

@@ -101,8 +101,13 @@ void shellInsertChar(SShellCmd *cmd, char *c, int32_t size) {
   /* update the values */
   cmd->commandSize += size;
   cmd->cursorOffset += size;
+  for (int i = 0; i < size; i++) {
+    taosMbToWchar(&wc, c + i, size);
     cmd->screenOffset += taosWcharWidth(wc);
     cmd->endOffset += taosWcharWidth(wc);
+  }
+  // set string end
+  cmd->command[cmd->commandSize] = 0;
 #ifdef WINDOWS
 #else
   shellShowOnScreen(cmd);
@@ -123,6 +128,8 @@ void shellBackspaceChar(SShellCmd *cmd) {
     cmd->cursorOffset -= size;
     cmd->screenOffset -= width;
     cmd->endOffset -= width;
+    // set string end
+    cmd->command[cmd->commandSize] = 0;
     shellShowOnScreen(cmd);
   }
 }
@@ -136,6 +143,8 @@ void shellClearLineBefore(SShellCmd *cmd) {
   cmd->cursorOffset = 0;
   cmd->screenOffset = 0;
   cmd->endOffset = cmd->commandSize;
+  // set string end
+  cmd->command[cmd->commandSize] = 0;
   shellShowOnScreen(cmd);
 }
 
@@ -160,6 +169,8 @@ void shellDeleteChar(SShellCmd *cmd) {
             cmd->commandSize - cmd->cursorOffset - size);
     cmd->commandSize -= size;
     cmd->endOffset -= width;
+    // set string end
+    cmd->command[cmd->commandSize] = 0;
     shellShowOnScreen(cmd);
   }
 }

tools/shell/src/shellTire.c

@@ -309,27 +309,24 @@ void matchPrefixFromTree(STire* tire, char* prefix, SMatch* match) {
 }
 
 SMatch* matchPrefix(STire* tire, char* prefix, SMatch* match) {
-  if (match == NULL) {
-    match = (SMatch*)taosMemoryMalloc(sizeof(SMatch));
-    memset(match, 0, sizeof(SMatch));
+  SMatch* rMatch = match; // define return match
+  if (rMatch == NULL) {
+    rMatch = (SMatch*)taosMemoryMalloc(sizeof(SMatch));
+    memset(rMatch, 0, sizeof(SMatch));
   }
 
   switch (tire->type) {
     case TIRE_TREE:
-      matchPrefixFromTree(tire, prefix, match);
+      matchPrefixFromTree(tire, prefix, rMatch);
+      break;
     case TIRE_LIST:
-      matchPrefixFromList(tire, prefix, match);
+      matchPrefixFromList(tire, prefix, rMatch);
+      break;
     default:
       break;
   }
 
-  // return if need
-  if (match->count == 0) {
-    freeMatch(match);
-    match = NULL;
-  }
-
-  return match;
+  return rMatch;
 }
 
 // get all items from tires tree
@@ -378,8 +375,10 @@ SMatch* enumAll(STire* tire) {
   switch (tire->type) {
     case TIRE_TREE:
       enumFromTree(tire, match);
+      break;
     case TIRE_LIST:
       enumFromList(tire, match);
+      break;
     default:
       break;
   }

utils/tsim/inc/simInt.h

@@ -19,20 +19,20 @@
 #include "os.h"
 
 #include "cJSON.h"
-#include "tconfig.h"
 #include "taos.h"
 #include "taoserror.h"
+#include "tconfig.h"
+#include "tglobal.h"
 #include "tidpool.h"
 #include "tlog.h"
 #include "ttimer.h"
 #include "ttypes.h"
 #include "tutil.h"
-#include "tglobal.h"
 
 #define MAX_MAIN_SCRIPT_NUM       10
 #define MAX_BACKGROUND_SCRIPT_NUM 10
 #define MAX_FILE_NAME_LEN         256
-#define MAX_ERROR_LEN             1024
+#define MAX_ERROR_LEN             4096
 #define MAX_QUERY_VALUE_LEN       1024
 #define MAX_QUERY_COL_NUM         100
 #define MAX_QUERY_ROW_NUM         100
@@ -55,12 +55,42 @@
 #define FAILED_POSTFIX  ""
 #endif
 
-#define simFatal(...) { if (simDebugFlag & DEBUG_FATAL) { taosPrintLog("SIM FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
-#define simError(...) { if (simDebugFlag & DEBUG_ERROR) { taosPrintLog("SIM ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
-#define simWarn(...)  { if (simDebugFlag & DEBUG_WARN)  { taosPrintLog("SIM WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
-#define simInfo(...)  { if (simDebugFlag & DEBUG_INFO)  { taosPrintLog("SIM ", DEBUG_INFO, 255, __VA_ARGS__); }}
-#define simDebug(...) { if (simDebugFlag & DEBUG_DEBUG) { taosPrintLog("SIM ", DEBUG_DEBUG, simDebugFlag, __VA_ARGS__); }}
-#define simTrace(...) { if (simDebugFlag & DEBUG_TRACE) { taosPrintLog("SIM ", DEBUG_TRACE, simDebugFlag, __VA_ARGS__); }}
+#define simFatal(...)                                            \
+  {                                                              \
+    if (simDebugFlag & DEBUG_FATAL) {                            \
+      taosPrintLog("SIM FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
+    }                                                            \
+  }
+#define simError(...)                                            \
+  {                                                              \
+    if (simDebugFlag & DEBUG_ERROR) {                            \
+      taosPrintLog("SIM ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
+    }                                                            \
+  }
+#define simWarn(...)                                           \
+  {                                                            \
+    if (simDebugFlag & DEBUG_WARN) {                           \
+      taosPrintLog("SIM WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
+    }                                                          \
+  }
+#define simInfo(...)                                      \
+  {                                                       \
+    if (simDebugFlag & DEBUG_INFO) {                      \
+      taosPrintLog("SIM ", DEBUG_INFO, 255, __VA_ARGS__); \
+    }                                                     \
+  }
+#define simDebug(...)                                               \
+  {                                                                 \
+    if (simDebugFlag & DEBUG_DEBUG) {                               \
+      taosPrintLog("SIM ", DEBUG_DEBUG, simDebugFlag, __VA_ARGS__); \
+    }                                                               \
+  }
+#define simTrace(...)                                               \
+  {                                                                 \
+    if (simDebugFlag & DEBUG_TRACE) {                               \
+      taosPrintLog("SIM ", DEBUG_TRACE, simDebugFlag, __VA_ARGS__); \
+    }                                                               \
+  }
 
 enum { SIM_SCRIPT_TYPE_MAIN, SIM_SCRIPT_TYPE_BACKGROUND };