Merge branch '3.0' into fix/TD-19254-D

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/common/ttypes.h

@@ -340,12 +340,8 @@ typedef struct tDataTypeDescriptor {
 } tDataTypeDescriptor;
 
 extern tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX];
-
 bool isValidDataType(int32_t type);
 
-void  setVardataNull(void *val, int32_t type);
-//void  setNull(void *val, int32_t type, int32_t bytes);
-//void  setNullN(void *val, int32_t type, int32_t bytes, int32_t numOfElems);
 void  assignVal(char *val, const char *src, int32_t len, int32_t type);
 void  operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type);
 void *getDataMin(int32_t type);

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);

include/util/tdef.h

@@ -237,7 +237,6 @@ typedef enum ELogicConditionType {
 #define TSDB_MAX_BYTES_PER_ROW  49151
 #define TSDB_MAX_TAGS_LEN       16384
 #define TSDB_MAX_TAGS           128
-#define TSDB_MAX_TAG_CONDITIONS 1024
 
 #define TSDB_MAX_COL_TAG_NUM  (TSDB_MAX_COLUMNS + TSDB_MAX_TAGS)
 #define TSDB_MAX_JSON_TAG_LEN 16384
@@ -274,8 +273,6 @@ typedef enum ELogicConditionType {
 
 #define TSDB_PAYLOAD_SIZE         TSDB_DEFAULT_PKT_SIZE
 #define TSDB_DEFAULT_PAYLOAD_SIZE 5120  // default payload size, greater than PATH_MAX value
-#define TSDB_EXTRA_PAYLOAD_SIZE   128   // extra bytes for auth
-#define TSDB_CQ_SQL_SIZE          1024
 #define TSDB_MIN_VNODES           16
 #define TSDB_MAX_VNODES           512
 
@@ -285,10 +282,7 @@ typedef enum ELogicConditionType {
 
 #define TSDB_MAX_REPLICA 5
 
-#define TSDB_TBNAME_COLUMN_INDEX (-1)
-#define TSDB_UD_COLUMN_INDEX     (-1000)
-#define TSDB_RES_COL_ID          (-5000)
-
+#define TSDB_TBNAME_COLUMN_INDEX     (-1)
 #define TSDB_MULTI_TABLEMETA_MAX_NUM 100000  // maximum batch size allowed to load table meta
 
 #define TSDB_MIN_VNODES_PER_DB          1
@@ -398,9 +392,6 @@ typedef enum ELogicConditionType {
 #define TSDB_MAX_EXPLAIN_RATIO     1
 #define TSDB_DEFAULT_EXPLAIN_RATIO 0.001
 
-#define TSDB_MAX_JOIN_TABLE_NUM 10
-#define TSDB_MAX_UNION_CLAUSE   5
-
 #define TSDB_DEFAULT_EXPLAIN_VERBOSE false
 
 #define TSDB_EXPLAIN_RESULT_ROW_SIZE    (16 * 1024)
@@ -419,7 +410,6 @@ typedef enum ELogicConditionType {
 #endif
 
 #define TSDB_QUERY_TYPE_NON_TYPE      0x00u  // none type
-#define TSDB_QUERY_TYPE_FREE_RESOURCE 0x01u  // free qhandle at vnode
 
 #define TSDB_META_COMPACT_RATIO 0  // disable tsdb meta compact by default
 

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/common/src/ttime.c

@@ -244,8 +244,7 @@ int32_t parseTimezone(char* str, int64_t* tzOffset) {
  * 2013-04-12T15:52:01.123+0800
  */
 int32_t parseTimeWithTz(const char* timestr, int64_t* time, int32_t timePrec, char delim) {
-  int64_t factor =
-      (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
+  int64_t factor = TSDB_TICK_PER_SECOND(timePrec);
   int64_t tzOffset = 0;
 
   struct tm tm = {0};
@@ -339,8 +338,8 @@ static FORCE_INLINE bool validateTm(struct tm* pTm) {
   return true;
 }
 
-int32_t parseLocaltime(char* timestr, int32_t len, int64_t* time, int32_t timePrec, char delim) {
-  *time = 0;
+int32_t parseLocaltime(char* timestr, int32_t len, int64_t* utime, int32_t timePrec, char delim) {
+  *utime = 0;
   struct tm tm = {0};
 
   char* str;
@@ -378,15 +377,12 @@ int32_t parseLocaltime(char* timestr, int32_t len, int64_t* time, int32_t timePr
     }
   }
 
-  int64_t factor =
-      (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
-  *time = factor * seconds + fraction;
-
+  *utime = TSDB_TICK_PER_SECOND(timePrec) * seconds + fraction;
   return 0;
 }
 
-int32_t parseLocaltimeDst(char* timestr, int32_t len, int64_t* time, int32_t timePrec, char delim) {
-  *time = 0;
+int32_t parseLocaltimeDst(char* timestr, int32_t len, int64_t* utime, int32_t timePrec, char delim) {
+  *utime = 0;
   struct tm tm = {0};
   tm.tm_isdst = -1;
 
@@ -411,7 +407,6 @@ int32_t parseLocaltimeDst(char* timestr, int32_t len, int64_t* time, int32_t tim
   int64_t seconds = taosMktime(&tm);
 
   int64_t fraction = 0;
-
   if (*str == '.') {
     /* parse the second fraction part */
     if ((fraction = parseFraction(str + 1, &str, timePrec)) < 0) {
@@ -419,9 +414,7 @@ int32_t parseLocaltimeDst(char* timestr, int32_t len, int64_t* time, int32_t tim
     }
   }
 
-  int64_t factor =
-      (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
-  *time = factor * seconds + fraction;
+  *utime = TSDB_TICK_PER_SECOND(timePrec) * seconds + fraction;
   return 0;
 }
 
@@ -437,58 +430,61 @@ char getPrecisionUnit(int32_t precision) {
   }
 }
 
-int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
-  assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
+int64_t convertTimePrecision(int64_t utime, int32_t fromPrecision, int32_t toPrecision) {
+  ASSERT(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
          fromPrecision == TSDB_TIME_PRECISION_NANO);
-  assert(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO ||
+  ASSERT(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO ||
          toPrecision == TSDB_TIME_PRECISION_NANO);
-  double tempResult = (double)time;
+
+  double tempResult = (double)utime;
+
   switch (fromPrecision) {
     case TSDB_TIME_PRECISION_MILLI: {
       switch (toPrecision) {
         case TSDB_TIME_PRECISION_MILLI:
-          return time;
+          return utime;
         case TSDB_TIME_PRECISION_MICRO:
           tempResult *= 1000;
-          time *= 1000;
+          utime *= 1000;
           goto end_;
         case TSDB_TIME_PRECISION_NANO:
           tempResult *= 1000000;
-          time *= 1000000;
+          utime *= 1000000;
           goto end_;
       }
     }  // end from milli
     case TSDB_TIME_PRECISION_MICRO: {
       switch (toPrecision) {
         case TSDB_TIME_PRECISION_MILLI:
-          return time / 1000;
+          return utime / 1000;
         case TSDB_TIME_PRECISION_MICRO:
-          return time;
+          return utime;
         case TSDB_TIME_PRECISION_NANO:
           tempResult *= 1000;
-          time *= 1000;
+          utime *= 1000;
           goto end_;
       }
     }  // end from micro
     case TSDB_TIME_PRECISION_NANO: {
       switch (toPrecision) {
         case TSDB_TIME_PRECISION_MILLI:
-          return time / 1000000;
+          return utime / 1000000;
         case TSDB_TIME_PRECISION_MICRO:
-          return time / 1000;
+          return utime / 1000;
         case TSDB_TIME_PRECISION_NANO:
-          return time;
+          return utime;
       }
     }  // end from nano
     default: {
       assert(0);
-      return time;  // only to pass windows compilation
+      return utime;  // only to pass windows compilation
     }
   }  // end switch fromPrecision
+
 end_:
   if (tempResult >= (double)INT64_MAX) return INT64_MAX;
   if (tempResult <= (double)INT64_MIN) return INT64_MIN;  // INT64_MIN means NULL
-  return time;
+  return utime;
 }
 
 // !!!!notice:there are precision problems, double lose precison if time is too large, for example:

source/common/src/ttypes.c

@@ -16,7 +16,6 @@
 #define _DEFAULT_SOURCE
 #include "ttypes.h"
 #include "tcompression.h"
-#include "trow.h"
 
 const int32_t TYPE_BYTES[16] = {
     -1,                      // TSDB_DATA_TYPE_NULL
@@ -86,18 +85,6 @@ FORCE_INLINE void *getDataMax(int32_t type) {
 
 bool isValidDataType(int32_t type) { return type >= TSDB_DATA_TYPE_NULL && type < TSDB_DATA_TYPE_MAX; }
 
-void setVardataNull(void *val, int32_t type) {
-  if (type == TSDB_DATA_TYPE_BINARY) {
-    varDataSetLen(val, sizeof(int8_t));
-    *(uint8_t *)varDataVal(val) = TSDB_DATA_BINARY_NULL;
-  } else if (type == TSDB_DATA_TYPE_NCHAR) {
-    varDataSetLen(val, sizeof(int32_t));
-    *(uint32_t *)varDataVal(val) = TSDB_DATA_NCHAR_NULL;
-  } else {
-    assert(0);
-  }
-}
-
 #define POINTER_SHIFT(p, b) ((void *)((char *)(p) + (b)))
 
 void assignVal(char *val, const char *src, int32_t len, int32_t type) {

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/dnode/vnode/src/meta/metaTable.c

@@ -575,7 +575,11 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
   tdbTbGet(pMeta->pTbDb, &(STbDbKey){.version = version, .uid = uid}, sizeof(STbDbKey), &pData, &nData);
 
   tDecoderInit(&dc, pData, nData);
-  metaDecodeEntry(&dc, &e);
+  rc = metaDecodeEntry(&dc, &e);
+  if (rc < 0) {
+    tDecoderClear(&dc);
+    return -1;
+  }
 
   if (type) *type = e.type;
 
@@ -668,6 +672,11 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
   SSchemaWrapper *pSchema;
   int             c;
 
+  if (pAlterTbReq->colName == NULL) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    return -1;
+  }
+
   // search name index
   ret = tdbTbGet(pMeta->pNameIdx, pAlterTbReq->tbName, strlen(pAlterTbReq->tbName) + 1, &pVal, &nVal);
   if (ret < 0) {
@@ -1244,7 +1253,10 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
   tdbTbGet(pMeta->pTbDb, &tbDbKey, sizeof(tbDbKey), &pData, &nData);
 
   tDecoderInit(&dc, pData, nData);
-  metaDecodeEntry(&dc, &stbEntry);
+  ret = metaDecodeEntry(&dc, &stbEntry);
+  if (ret < 0) {
+    goto end;
+  }
 
   pTagColumn = &stbEntry.stbEntry.schemaTag.pSchema[0];
 

source/dnode/vnode/src/tsdb/tsdbCacheRead.c

@@ -194,7 +194,7 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
     }
   } else if ((pr->type & CACHESCAN_RETRIEVE_TYPE_ALL) == CACHESCAN_RETRIEVE_TYPE_ALL) {
     for (int32_t i = pr->tableIndex; i < numOfTables; ++i) {
-      STableKeyInfo* pKeyInfo = taosArrayGet(pr->pTableList, i);
+      STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(pr->pTableList, i);
       code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h);
       if (code != TSDB_CODE_SUCCESS) {
         return code;

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -1054,6 +1054,7 @@ static int32_t tsdbEndCommit(SCommitter *pCommitter, int32_t eno) {
 _exit:
   tsdbFSDestroy(&pCommitter->fs);
   taosArrayDestroy(pCommitter->aTbDataP);
+  pCommitter->aTbDataP = NULL;
   if (code || eno) {
     tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
   } else {

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -149,6 +149,7 @@ _exit:
 void tsdbFSDestroy(STsdbFS *pFS) {
   if (pFS->pDelFile) {
     taosMemoryFree(pFS->pDelFile);
+    pFS->pDelFile = NULL;
   }
 
   for (int32_t iSet = 0; iSet < taosArrayGetSize(pFS->aDFileSet); iSet++) {
@@ -162,6 +163,7 @@ void tsdbFSDestroy(STsdbFS *pFS) {
   }
 
   taosArrayDestroy(pFS->aDFileSet);
+  pFS->aDFileSet = NULL;
 }
 
 static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
@@ -296,7 +298,7 @@ static int32_t tsdbLoadFSFromFile(const char *fname, STsdbFS *pFS) {
   // load binary
   TdFilePtr pFD = taosOpenFile(fname, TD_FILE_READ);
   if (pFD == NULL) {
-    code = terrno;
+    code = TAOS_SYSTEM_ERROR(errno);
     TSDB_CHECK_CODE(code, lino, _exit);
   }
 

source/dnode/vnode/src/tsdb/tsdbMergeTree.c

@@ -120,44 +120,46 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
     return &pInfo->blockData[1];
   }
 
-  pInfo->currentLoadBlockIndex ^= 1;
-  if (pIter->pSttBlk != NULL) {  // current block not loaded yet
-    int64_t st = taosGetTimestampUs();
+  if (pIter->pSttBlk == NULL) {
+    return NULL;
+  }
 
-    SBlockData *pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex];
+  // current block not loaded yet
+  pInfo->currentLoadBlockIndex ^= 1;
+  int64_t st = taosGetTimestampUs();
 
-    TABLEID id = {0};
-    if (pIter->pSttBlk->suid != 0) {
-      id.suid = pIter->pSttBlk->suid;
-    } else {
-      id.uid = pIter->uid;
-    }
+  SBlockData *pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex];
 
-    code = tBlockDataInit(pBlock, &id, pInfo->pSchema, pInfo->colIds, pInfo->numOfCols);
-    if (code != TSDB_CODE_SUCCESS) {
-      goto _exit;
-    }
+  TABLEID id = {0};
+  if (pIter->pSttBlk->suid != 0) {
+    id.suid = pIter->pSttBlk->suid;
+  } else {
+    id.uid = pIter->uid;
+  }
 
-    code = tsdbReadSttBlock(pIter->pReader, pIter->iStt, pIter->pSttBlk, pBlock);
-    if (code != TSDB_CODE_SUCCESS) {
-      goto _exit;
-    }
+  code = tBlockDataInit(pBlock, &id, pInfo->pSchema, pInfo->colIds, pInfo->numOfCols);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _exit;
+  }
 
-    double el = (taosGetTimestampUs() - st) / 1000.0;
-    pInfo->elapsedTime += el;
-    pInfo->loadBlocks += 1;
+  code = tsdbReadSttBlock(pIter->pReader, pIter->iStt, pIter->pSttBlk, pBlock);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _exit;
+  }
 
-    tsdbDebug("read last block, total load:%d, trigger by uid:%" PRIu64
-              ", last file index:%d, last block index:%d, entry:%d, %p, elapsed time:%.2f ms, %s",
-              pInfo->loadBlocks, pIter->uid, pIter->iStt, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock, el,
-              idStr);
+  double el = (taosGetTimestampUs() - st) / 1000.0;
+  pInfo->elapsedTime += el;
+  pInfo->loadBlocks += 1;
 
-    pInfo->blockIndex[pInfo->currentLoadBlockIndex] = pIter->iSttBlk;
-    tsdbDebug("last block index list:%d, %d, %s", pInfo->blockIndex[0], pInfo->blockIndex[1], idStr);
+  tsdbDebug("read last block, total load:%d, trigger by uid:%" PRIu64
+            ", last file index:%d, last block index:%d, entry:%d, %p, elapsed time:%.2f ms, %s",
+            pInfo->loadBlocks, pIter->uid, pIter->iStt, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock, el,
+            idStr);
 
-    pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].nRow : -1;
-  }
+  pInfo->blockIndex[pInfo->currentLoadBlockIndex] = pIter->iSttBlk;
+  tsdbDebug("last block index list:%d, %d, %s", pInfo->blockIndex[0], pInfo->blockIndex[1], idStr);
 
+  pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].nRow : -1;
   return &pInfo->blockData[pInfo->currentLoadBlockIndex];
 
 _exit:
@@ -259,7 +261,8 @@ static int32_t binarySearchForStartRowIndex(uint64_t *uidList, int32_t num, uint
 int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t iStt, int8_t backward, uint64_t suid,
                        uint64_t uid, STimeWindow *pTimeWindow, SVersionRange *pRange, SSttBlockLoadInfo *pBlockLoadInfo,
                        const char *idStr) {
-  int32_t code = 0;
+  int32_t code = TSDB_CODE_SUCCESS;
+
   *pIter = taosMemoryCalloc(1, sizeof(SLDataIter));
   if (*pIter == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
@@ -336,7 +339,10 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
     (*pIter)->iRow = ((*pIter)->backward) ? (*pIter)->pSttBlk->nRow : -1;
   }
 
+  return code;
+
 _exit:
+  taosMemoryFree(*pIter);
   return code;
 }
 
@@ -473,7 +479,7 @@ bool tLDataIterNextRow(SLDataIter *pIter, const char *idStr) {
 
   int32_t     iBlockL = pIter->iSttBlk;
   SBlockData *pBlockData = loadLastBlock(pIter, idStr);
-  if (pBlockData == NULL && terrno != TSDB_CODE_SUCCESS) {
+  if (pBlockData == NULL || terrno != TSDB_CODE_SUCCESS) {
     goto _exit;
   }
 

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -1047,11 +1047,16 @@ static int32_t fileDataBlockOrderCompar(const void* pLeft, const void* pRight, v
   return pLeftBlock->offset > pRightBlock->offset ? 1 : -1;
 }
 
-static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter) {
+static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter, const char* idStr) {
   SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
   if (pBlockInfo != NULL) {
     STableBlockScanInfo* pScanInfo = taosHashGet(pBlockIter->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
-    int32_t*             mapDataIndex = taosArrayGet(pScanInfo->pBlockList, pBlockInfo->tbBlockIdx);
+    if (pScanInfo == NULL) {
+      tsdbError("failed to locate the uid:%"PRIu64" in query table uid list, %s", pBlockInfo->uid, idStr);
+      return TSDB_CODE_INVALID_PARA;
+    }
+
+    int32_t* mapDataIndex = taosArrayGet(pScanInfo->pBlockList, pBlockInfo->tbBlockIdx);
     tMapDataGetItemByIdx(&pScanInfo->mapData, *mapDataIndex, &pBlockIter->block, tGetDataBlk);
   }
 
@@ -1135,7 +1140,7 @@ static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIte
 
     pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
     cleanupBlockOrderSupporter(&sup);
-    doSetCurrentBlock(pBlockIter);
+    doSetCurrentBlock(pBlockIter, pReader->idStr);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1175,12 +1180,12 @@ static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIte
   taosMemoryFree(pTree);
 
   pBlockIter->index = asc ? 0 : (numOfBlocks - 1);
-  doSetCurrentBlock(pBlockIter);
+  doSetCurrentBlock(pBlockIter, pReader->idStr);
 
   return TSDB_CODE_SUCCESS;
 }
 
-static bool blockIteratorNext(SDataBlockIter* pBlockIter) {
+static bool blockIteratorNext(SDataBlockIter* pBlockIter, const char* idStr) {
   bool asc = ASCENDING_TRAVERSE(pBlockIter->order);
 
   int32_t step = asc ? 1 : -1;
@@ -1189,7 +1194,7 @@ static bool blockIteratorNext(SDataBlockIter* pBlockIter) {
   }
 
   pBlockIter->index += step;
-  doSetCurrentBlock(pBlockIter);
+  doSetCurrentBlock(pBlockIter, idStr);
 
   return true;
 }
@@ -1260,7 +1265,7 @@ static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t
     ASSERT(pBlockInfo->uid == fblock.uid && pBlockInfo->tbBlockIdx == fblock.tbBlockIdx);
   }
 
-  doSetCurrentBlock(pBlockIter);
+  doSetCurrentBlock(pBlockIter, "");
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2190,6 +2195,8 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI
 }
 
 static int32_t buildComposedDataBlock(STsdbReader* pReader) {
+  int32_t code = TSDB_CODE_SUCCESS;
+
   SSDataBlock* pResBlock = pReader->pResBlock;
 
   SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
@@ -2200,6 +2207,13 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
   STableBlockScanInfo* pBlockScanInfo = NULL;
   if (pBlockInfo != NULL) {
     pBlockScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
+    if (pBlockScanInfo == NULL) {
+      code = TSDB_CODE_INVALID_PARA;
+      tsdbError("failed to locate the uid:%"PRIu64" in query table uid list, total tables:%d, %s",
+          pBlockInfo->uid, taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
+      goto _end;
+    }
+
     SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter);
     TSDBKEY   keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader);
 
@@ -2276,7 +2290,7 @@ _end:
               pResBlock->info.rows, el, pReader->idStr);
   }
 
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
 
 void setComposedBlockFlag(STsdbReader* pReader, bool composed) { pReader->status.composedDataBlock = composed; }
@@ -2732,7 +2746,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
       // current block are exhausted, try the next file block
       if (pDumpInfo->allDumped) {
         // try next data block in current file
-        bool hasNext = blockIteratorNext(&pReader->status.blockIter);
+        bool hasNext = blockIteratorNext(&pReader->status.blockIter, pReader->idStr);
         if (hasNext) {  // check for the next block in the block accessed order list
           initBlockDumpInfo(pReader, pBlockIter);
         } else {
@@ -3658,10 +3672,6 @@ void tsdbReaderClose(STsdbReader* pReader) {
     taosMemoryFree(pLReader);
   }
 
-  if (pReader->innerReader[0] != 0) {
-    tsdbUntakeReadSnap(pReader->innerReader[0]->pTsdb, pReader->innerReader[0]->pReadSnap, pReader->idStr);
-  }
-
   tsdbDebug(
       "%p :io-cost summary: head-file:%" PRIu64 ", head-file time:%.2f ms, SMA:%" PRId64
       " SMA-time:%.2f ms, fileBlocks:%" PRId64
@@ -3849,8 +3859,14 @@ static SArray* doRetrieveDataBlock(STsdbReader* pReader) {
     return pReader->pResBlock->pDataBlock;
   }
 
-  SFileDataBlockInfo*  pFBlock = getCurrentBlockInfo(&pStatus->blockIter);
-  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
+  SFileDataBlockInfo*  pBlockInfo = getCurrentBlockInfo(&pStatus->blockIter);
+  STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
+  if (pBlockScanInfo == NULL) {
+    terrno = TSDB_CODE_INVALID_PARA;
+    tsdbError("failed to locate the uid:%" PRIu64 " in query table uid list, total tables:%d, %s", pBlockInfo->uid,
+              taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
+    return NULL;
+  }
 
   int32_t code = doLoadFileBlockData(pReader, &pStatus->blockIter, &pStatus->fileBlockData, pBlockScanInfo->uid);
   if (code != TSDB_CODE_SUCCESS) {
@@ -3979,7 +3995,7 @@ int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTa
       int32_t bucketIndex = getBucketIndex(pTableBlockInfo->defMinRows, bucketRange, numOfRows);
       pTableBlockInfo->blockRowsHisto[bucketIndex]++;
 
-      hasNext = blockIteratorNext(&pStatus->blockIter);
+      hasNext = blockIteratorNext(&pStatus->blockIter, pReader->idStr);
     } else {
       code = initForFirstBlockInFile(pReader, pBlockIter);
       if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {

source/dnode/vnode/src/tsdb/tsdbRetention.c

@@ -47,7 +47,7 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
   }
 
   // do retention
-  STsdbFS fs;
+  STsdbFS fs = {0};
 
   code = tsdbFSCopy(pTsdb, &fs);
   if (code) goto _err;

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -285,15 +285,43 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   }
 
   for (int32_t i = 0; i < msgNum; ++i) {
+    if (offset >= pMsg->contLen) {
+      qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      taosArrayDestroy(batchRsp);
+      return -1;
+    }    
+
     req.msgIdx = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
     offset += sizeof(req.msgIdx);
 
+    if (offset >= pMsg->contLen) {
+      qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      taosArrayDestroy(batchRsp);
+      return -1;
+    } 
+
     req.msgType = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
     offset += sizeof(req.msgType);
 
+    if (offset >= pMsg->contLen) {
+      qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      taosArrayDestroy(batchRsp);
+      return -1;
+    } 
+
     req.msgLen = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
     offset += sizeof(req.msgLen);
 
+    if (offset >= pMsg->contLen) {
+      qError("vnode offset %d is bigger than contLen %d", offset, pMsg->contLen);
+      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+      taosArrayDestroy(batchRsp);
+      return -1;
+    } 
+
     req.msg = (char *)pMsg->pCont + offset;
     offset += req.msgLen;
 

source/libs/executor/inc/executorimpl.h

@@ -954,9 +954,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
 SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode* pScanPhyNode,
                                               const char* pUser, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
-                                           SSDataBlock* pResultBlock, SNode* pCondition, SExprInfo* pScalarExprInfo,
-                                           int32_t numOfScalarExpr, bool mergeResult, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pNode, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pNode,
                                                  SExecTaskInfo* pTaskInfo);
@@ -980,9 +978,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
                                                      SExecTaskInfo* pTaskInfo, int32_t numOfChild);
 SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
                                             SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
-                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
-                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode *pAggNode, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SReadHandle* readHandle, uint64_t uid,
                                                SBlockDistScanPhysiNode* pBlockScanNode, SExecTaskInfo* pTaskInfo);
 

source/libs/executor/src/cachescanoperator.c

@@ -162,7 +162,9 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
 
       if (pTableList->map != NULL) {
         int64_t* groupId = taosHashGet(pTableList->map, &pInfo->pRes->info.uid, sizeof(int64_t));
-        pInfo->pRes->info.groupId = *groupId;
+        if (groupId != NULL) {
+          pInfo->pRes->info.groupId = *groupId;
+        }
       } else {
         ASSERT(taosArrayGetSize(pTableList->pTableList) == 1);
         STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, 0);

source/libs/executor/src/dataDeleter.c

@@ -234,8 +234,11 @@ static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
   while (!taosQueueEmpty(pDeleter->pDataBlocks)) {
     SDataDeleterBuf* pBuf = NULL;
     taosReadQitem(pDeleter->pDataBlocks, (void**)&pBuf);
-    taosMemoryFreeClear(pBuf->pData);
-    taosFreeQitem(pBuf);
+
+    if (pBuf != NULL) {
+      taosMemoryFreeClear(pBuf->pData);
+      taosFreeQitem(pBuf);
+    }
   }
   taosCloseQueue(pDeleter->pDataBlocks);
   taosThreadMutexDestroy(&pDeleter->mutex);

source/libs/executor/src/dataInserter.c

@@ -329,6 +329,8 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
   }
 
   if (pInserterNode->stableId != suid) {
+    destroyDataSinker((SDataSinkHandle*)inserter);
+    taosMemoryFree(inserter);
     terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
     return terrno;
   }
@@ -336,6 +338,8 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
   inserter->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
   taosThreadMutexInit(&inserter->mutex, NULL);
   if (NULL == inserter->pDataBlocks) {
+    destroyDataSinker((SDataSinkHandle*)inserter);
+    taosMemoryFree(inserter);
     terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }

source/libs/executor/src/executor.c

@@ -358,7 +358,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
   SDataSinkMgtCfg cfg = {.maxDataBlockNum = 10000, .maxDataBlockNumPerQuery = 5000};
   code = dsDataSinkMgtInit(&cfg);
   if (code != TSDB_CODE_SUCCESS) {
-    qError("failed to dsDataSinkMgtInit, code: %s", tstrerror(code));
+    qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str);
     goto _error;
   }
 
@@ -366,7 +366,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
     void* pSinkParam = NULL;
     code = createDataSinkParam(pSubplan->pDataSink, &pSinkParam, pTaskInfo, readHandle);
     if (code != TSDB_CODE_SUCCESS) {
-      qError("failed to createDataSinkParam, code: %s", tstrerror(code));
+      qError("failed to createDataSinkParam, vgId:%d, code:%s, %s", vgId, tstrerror(code), (*pTask)->id.str);
       goto _error;
     }
 

source/libs/executor/src/executorimpl.c

@@ -3048,32 +3048,40 @@ void cleanupExprSupp(SExprSupp* pSupp) {
   taosMemoryFree(pSupp->rowEntryInfoOffset);
 }
 
-SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
-                                           SSDataBlock* pResultBlock, SNode* pCondition, SExprInfo* pScalarExprInfo,
-                                           int32_t numOfScalarExpr, bool mergeResult, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode,SExecTaskInfo* pTaskInfo) {
   SAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SAggOperatorInfo));
   SOperatorInfo*    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
+  SSDataBlock* pResBlock = createResDataBlock(pAggNode->node.pOutputDataBlockDesc);
+  initBasicInfo(&pInfo->binfo, pResBlock);
+
+  int32_t    numOfScalarExpr = 0;
+  SExprInfo* pScalarExprInfo = NULL;
+  if (pAggNode->pExprs != NULL) {
+    pScalarExprInfo = createExprInfo(pAggNode->pExprs, NULL, &numOfScalarExpr);
+  }
 
+  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
   initResultSizeInfo(&pOperator->resultInfo, 4096);
-  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
+
+  int32_t      num = 0;
+  SExprInfo*   pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num);
+  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  initBasicInfo(&pInfo->binfo, pResultBlock);
   code = initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  pInfo->binfo.mergeResultBlock = mergeResult;
+  pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
   pInfo->groupId = UINT64_MAX;
-  pInfo->pCondition = pCondition;
+  pInfo->pCondition = pAggNode->node.pConditions;
   pOperator->name = "TableAggregate";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_AGG;
   pOperator->blocking = true;
@@ -3332,8 +3340,6 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPT
   return pTaskInfo;
 }
 
-static SArray* extractColumnInfo(SNodeList* pNodeList);
-
 SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode);
 
 int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, SExecTaskInfo* pTaskInfo) {
@@ -3710,22 +3716,10 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
     pOptr = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo);
   } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
     SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
-    SExprInfo*     pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num);
-    SSDataBlock*   pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
-
-    int32_t    numOfScalarExpr = 0;
-    SExprInfo* pScalarExprInfo = NULL;
-    if (pAggNode->pExprs != NULL) {
-      pScalarExprInfo = createExprInfo(pAggNode->pExprs, NULL, &numOfScalarExpr);
-    }
-
     if (pAggNode->pGroupKeys != NULL) {
-      SArray* pColList = extractColumnInfo(pAggNode->pGroupKeys);
-      pOptr = createGroupOperatorInfo(ops[0], pExprInfo, num, pResBlock, pColList, pAggNode->node.pConditions,
-                                      pScalarExprInfo, numOfScalarExpr, pTaskInfo);
+      pOptr = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo);
     } else {
-      pOptr = createAggregateOperatorInfo(ops[0], pExprInfo, num, pResBlock, pAggNode->node.pConditions,
-                                          pScalarExprInfo, numOfScalarExpr, pAggNode->mergeDataBlock, pTaskInfo);
+      pOptr = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo);
     }
   } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
     SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
@@ -3815,39 +3809,6 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
   return pOptr;
 }
 
-SArray* extractColumnInfo(SNodeList* pNodeList) {
-  size_t  numOfCols = LIST_LENGTH(pNodeList);
-  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
-  if (pList == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-
-  for (int32_t i = 0; i < numOfCols; ++i) {
-    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
-
-    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
-      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
-
-      SColumn c = extractColumnFromColumnNode(pColNode);
-      taosArrayPush(pList, &c);
-    } else if (nodeType(pNode->pExpr) == QUERY_NODE_VALUE) {
-      SValueNode* pValNode = (SValueNode*)pNode->pExpr;
-      SColumn     c = {0};
-      c.slotId = pNode->slotId;
-      c.colId = pNode->slotId;
-      c.type = pValNode->node.type;
-      c.bytes = pValNode->node.resType.bytes;
-      c.scale = pValNode->node.resType.scale;
-      c.precision = pValNode->node.resType.precision;
-
-      taosArrayPush(pList, &c);
-    }
-  }
-
-  return pList;
-}
-
 static int32_t extractTbscanInStreamOpTree(SOperatorInfo* pOperator, STableScanInfo** ppInfo) {
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     if (pOperator->numOfDownstream == 0) {
@@ -4071,6 +4032,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);

source/libs/executor/src/groupoperator.c

@@ -30,6 +30,7 @@ static void*    getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDa
 static int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity);
 static int32_t  setGroupResultOutputBuf(SOperatorInfo* pOperator, SOptrBasicInfo* binfo, int32_t numOfCols, char* pData,
                                         int16_t bytes, uint64_t groupId, SDiskbasedBuf* pBuf, SAggSupporter* pAggSup);
+static SArray*  extractColumnInfo(SNodeList* pNodeList);
 
 static void freeGroupKey(void* param) {
   SGroupKeys* pKey = (SGroupKeys*)param;
@@ -61,7 +62,7 @@ static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char**
 
   int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);
   for (int32_t i = 0; i < numOfGroupCols; ++i) {
-    SColumn* pCol = taosArrayGet(pGroupColList, i);
+    SColumn* pCol = (SColumn*) taosArrayGet(pGroupColList, i);
     (*keyLen) += pCol->bytes;  // actual data + null_flag
 
     SGroupKeys key = {0};
@@ -396,41 +397,48 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
   return buildGroupResultDataBlock(pOperator);
 }
 
-SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
-                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
-                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode *pAggNode, SExecTaskInfo* pTaskInfo) {
   SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  pInfo->pGroupCols = pGroupColList;
-  pInfo->pCondition = pCondition;
+  SSDataBlock* pResBlock = createResDataBlock(pAggNode->node.pOutputDataBlockDesc);
+  initBasicInfo(&pInfo->binfo, pResBlock);
+
+  int32_t    numOfScalarExpr = 0;
+  SExprInfo* pScalarExprInfo = NULL;
+  if (pAggNode->pExprs != NULL) {
+    pScalarExprInfo = createExprInfo(pAggNode->pExprs, NULL, &numOfScalarExpr);
+  }
+
+  pInfo->pGroupCols = extractColumnInfo(pAggNode->pGroupKeys);
+  pInfo->pCondition = pAggNode->node.pConditions;
 
   int32_t code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
+  int32_t    num = 0;
+  SExprInfo* pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num);
+  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pInfo->pGroupCols);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
   initResultSizeInfo(&pOperator->resultInfo, 4096);
-  code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pInfo->groupKeyLen, pTaskInfo->id.str);
+  code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, pInfo->groupKeyLen, pTaskInfo->id.str);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  initBasicInfo(&pInfo->binfo, pResultBlock);
   initResultRowInfo(&pInfo->binfo.resultRowInfo);
 
   pOperator->name = "GroupbyAggOperator";
   pOperator->blocking = true;
   pOperator->status = OP_NOT_OPENED;
-  // pOperator->operatorType = OP_Groupby;
   pOperator->info = pInfo;
   pOperator->pTaskInfo = pTaskInfo;
 
@@ -451,8 +459,6 @@ _error:
 }
 
 static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
-  //  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
-
   SPartitionOperatorInfo* pInfo = pOperator->info;
 
   for (int32_t j = 0; j < pBlock->info.rows; ++j) {
@@ -760,7 +766,6 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
 
   int32_t    numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);
-
   pInfo->pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);
 
   if (pPartNode->pExprs != NULL) {
@@ -781,14 +786,13 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
   uint32_t defaultPgsz = 0;
   uint32_t defaultBufsz = 0;
 
-  SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
-  getBufferPgSize(pResBlock->info.rowSize, &defaultPgsz, &defaultBufsz);
+  pInfo->binfo.pRes = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
+  getBufferPgSize(pInfo->binfo.pRes->info.rowSize, &defaultPgsz, &defaultBufsz);
 
   if (!osTempSpaceAvailable()) {
     terrno = TSDB_CODE_NO_AVAIL_DISK;
     pTaskInfo->code = terrno;
     qError("Create partition operator info failed since %s", terrstr(terrno));
-    blockDataDestroy(pResBlock);
     goto _error;
   }
 
@@ -797,8 +801,8 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
     goto _error;
   }
 
-  pInfo->rowCapacity = blockDataGetCapacityInRow(pResBlock, getBufPageSize(pInfo->pBuf));
-  pInfo->columnOffset = setupColumnOffset(pResBlock, pInfo->rowCapacity);
+  pInfo->rowCapacity = blockDataGetCapacityInRow(pInfo->binfo.pRes, getBufPageSize(pInfo->pBuf));
+  pInfo->columnOffset = setupColumnOffset(pInfo->binfo.pRes, pInfo->rowCapacity);
   code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pInfo->pGroupCols);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
@@ -808,7 +812,6 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
   pOperator->blocking = true;
   pOperator->status = OP_NOT_OPENED;
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION;
-  pInfo->binfo.pRes = pResBlock;
   pOperator->exprSupp.numOfExprs = numOfCols;
   pOperator->exprSupp.pExprInfo = pExprInfo;
   pOperator->info = pInfo;
@@ -1102,3 +1105,37 @@ _error:
   taosMemoryFreeClear(pOperator);
   return NULL;
 }
+
+
+SArray* extractColumnInfo(SNodeList* pNodeList) {
+  size_t  numOfCols = LIST_LENGTH(pNodeList);
+  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
+  if (pList == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);
+
+    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
+      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
+
+      SColumn c = extractColumnFromColumnNode(pColNode);
+      taosArrayPush(pList, &c);
+    } else if (nodeType(pNode->pExpr) == QUERY_NODE_VALUE) {
+      SValueNode* pValNode = (SValueNode*)pNode->pExpr;
+      SColumn     c = {0};
+      c.slotId = pNode->slotId;
+      c.colId = pNode->slotId;
+      c.type = pValNode->node.type;
+      c.bytes = pValNode->node.resType.bytes;
+      c.scale = pValNode->node.resType.scale;
+      c.precision = pValNode->node.resType.precision;
+
+      taosArrayPush(pList, &c);
+    }
+  }
+
+  return pList;
+}

source/libs/executor/src/joinoperator.c

@@ -59,15 +59,16 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
                                            SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo) {
   SJoinOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SJoinOperatorInfo));
   SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
+
+  int32_t code = TSDB_CODE_SUCCESS;
   if (pOperator == NULL || pInfo == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
     goto _error;
   }
 
+  int32_t      numOfCols = 0;
   SSDataBlock* pResBlock = createResDataBlock(pJoinNode->node.pOutputDataBlockDesc);
-
-  int32_t    numOfCols = 0;
-  SExprInfo* pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &numOfCols);
-
+  SExprInfo*   pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &numOfCols);
   initResultSizeInfo(&pOperator->resultInfo, 4096);
 
   pInfo->pRes = pResBlock;
@@ -84,8 +85,18 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
 
   if (pJoinNode->pOnConditions != NULL && pJoinNode->node.pConditions != NULL) {
     pInfo->pCondAfterMerge = nodesMakeNode(QUERY_NODE_LOGIC_CONDITION);
+    if (pInfo->pCondAfterMerge == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto _error;
+    }
+
     SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(pInfo->pCondAfterMerge);
     pLogicCond->pParameterList = nodesMakeList();
+    if (pLogicCond->pParameterList == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto _error;
+    }
+
     nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->pOnConditions));
     nodesListMakeAppend(&pLogicCond->pParameterList, nodesCloneNode(pJoinNode->node.pConditions));
     pLogicCond->condType = LOGIC_COND_TYPE_AND;
@@ -106,7 +117,7 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
 
   pOperator->fpSet =
       createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, NULL, destroyMergeJoinOperator, NULL, NULL, NULL);
-  int32_t code = appendDownstream(pOperator, pDownstream, numOfDownstream);
+  code = appendDownstream(pOperator, pDownstream, numOfDownstream);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -114,9 +125,12 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
   return pOperator;
 
 _error:
-  taosMemoryFree(pInfo);
+  if (pInfo != NULL) {
+    destroyMergeJoinOperator(pInfo);
+  }
+
   taosMemoryFree(pOperator);
-  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
+  pTaskInfo->code = code;
   return NULL;
 }
 

source/libs/executor/src/scanoperator.c

@@ -494,7 +494,13 @@ void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* p
   SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .param = pMeta, .columnData = &infoData};
 
   SScalarParam param = {.columnData = pColInfoData};
-  fpSet.process(&srcParam, 1, &param);
+
+  if (fpSet.process != NULL) {
+    fpSet.process(&srcParam, 1, &param);
+  } else {
+    qError("failed to get the corresponding callback function, functionId:%d", functionId);
+  }
+
   colDataDestroy(&infoData);
 }
 

source/libs/executor/src/sortoperator.c

@@ -719,12 +719,16 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   SOperatorInfo*              pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   SDataBlockDescNode*         pDescNode = pPhyNode->pOutputDataBlockDesc;
 
-  pInfo->binfo.pRes = createResDataBlock(pDescNode);
-  int32_t rowSize = pInfo->binfo.pRes->info.rowSize;
-  if (pInfo == NULL || pOperator == NULL || rowSize > 100 * 1024 * 1024) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (pInfo == NULL || pOperator == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
     goto _error;
   }
 
+  pInfo->binfo.pRes = createResDataBlock(pDescNode);
+  int32_t rowSize = pInfo->binfo.pRes->info.rowSize;
+  ASSERT(rowSize < 100 * 1024 * 1024);
+
   SArray* pSortInfo = createSortInfo(pMergePhyNode->pMergeKeys);
   int32_t numOfOutputCols = 0;
   SArray* pColMatchColInfo =
@@ -737,6 +741,9 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   pInfo->pSortInfo = pSortInfo;
   pInfo->pColMatchInfo = pColMatchColInfo;
   pInfo->pInputBlock = pInputBlock;
+  pInfo->bufPageSize = getProperSortPageSize(rowSize);
+  pInfo->sortBufSize = pInfo->bufPageSize * (numStreams + 1);  // one additional is reserved for merged result.
+
   pOperator->name = "MultiwayMerge";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE;
   pOperator->blocking = false;
@@ -744,15 +751,10 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   pOperator->info = pInfo;
   pOperator->pTaskInfo = pTaskInfo;
 
-  pInfo->bufPageSize = getProperSortPageSize(rowSize);
-
-  // one additional is reserved for merged result.
-  pInfo->sortBufSize = pInfo->bufPageSize * (numStreams + 1);
-
   pOperator->fpSet = createOperatorFpSet(doOpenMultiwayMergeOperator, doMultiwayMerge, NULL, NULL,
                                          destroyMultiwayMergeOperatorInfo, NULL, NULL, getMultiwayMergeExplainExecInfo);
 
-  int32_t code = appendDownstream(pOperator, downStreams, numStreams);
+  code = appendDownstream(pOperator, downStreams, numStreams);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }

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/executor/src/tlinearhash.c

@@ -251,6 +251,7 @@ SLHashObj* tHashInit(int32_t inMemPages, int32_t pageSize, _hash_fn_t fn, int32_
   if (!osTempSpaceAvailable()) {
     terrno = TSDB_CODE_NO_AVAIL_DISK;
     printf("tHash Init failed since %s", terrstr(terrno));
+    taosMemoryFree(pHashObj);
     return NULL;
   }
 

source/libs/executor/src/tsort.c

@@ -140,6 +140,7 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf* pBuf, SArray* pAllSource
                                          int32_t* sourceId, SArray* pPageIdList) {
   SSortSource* pSource = taosMemoryCalloc(1, sizeof(SSortSource));
   if (pSource == NULL) {
+    taosArrayDestroy(pPageIdList);
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
@@ -155,6 +156,7 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf* pBuf, SArray* pAllSource
   int32_t numOfRows =
       (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock))) / rowSize;
   ASSERT(numOfRows > 0);
+
   return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows);
 }
 
@@ -224,6 +226,22 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
 
   int32_t code = 0;
 
+  // multi-pass internal merge sort is required
+  if (pHandle->pBuf == NULL) {
+    if (!osTempSpaceAvailable()) {
+      code = TSDB_CODE_NO_AVAIL_DISK;
+      qError("Sort compare init failed since %s", terrstr(code));
+      return code;
+    }
+
+    code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize,
+                              "sortComparInit", tsTempDir);
+    dBufSetPrintInfo(pHandle->pBuf);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
+  }
+
   if (pHandle->type == SORT_SINGLESOURCE_SORT) {
     for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
       SSortSource* pSource = cmpParam->pSources[i];
@@ -245,22 +263,6 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
       releaseBufPage(pHandle->pBuf, pPage);
     }
   } else {
-    // multi-pass internal merge sort is required
-    if (pHandle->pBuf == NULL) {
-      if (!osTempSpaceAvailable()) {
-        terrno = TSDB_CODE_NO_AVAIL_DISK;
-        code = terrno;
-        qError("Sort compare init failed since %s", terrstr(terrno));
-        return code;
-      }
-      code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize,
-                                "sortComparInit", tsTempDir);
-      dBufSetPrintInfo(pHandle->pBuf);
-      if (code != TSDB_CODE_SUCCESS) {
-        return code;
-      }
-    }
-
     for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
       SSortSource* pSource = cmpParam->pSources[i];
       pSource->src.pBlock = pHandle->fetchfp(pSource->param);
@@ -507,12 +509,14 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
 
       int32_t code = sortComparInit(&pHandle->cmpParam, pHandle->pOrderedSource, i * numOfInputSources, end, pHandle);
       if (code != TSDB_CODE_SUCCESS) {
+        taosArrayDestroy(pResList);
         return code;
       }
 
       code =
           tMergeTreeCreate(&pHandle->pMergeTree, pHandle->cmpParam.numOfSources, &pHandle->cmpParam, pHandle->comparFn);
       if (code != TSDB_CODE_SUCCESS) {
+        taosArrayDestroy(pResList);
         return code;
       }
 
@@ -520,12 +524,16 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
       while (1) {
         SSDataBlock* pDataBlock = getSortedBlockDataInner(pHandle, &pHandle->cmpParam, numOfRows);
         if (pDataBlock == NULL) {
+          taosArrayDestroy(pResList);
+          taosArrayDestroy(pPageIdList);
           break;
         }
 
         int32_t pageId = -1;
         void*   pPage = getNewBufPage(pHandle->pBuf, &pageId);
         if (pPage == NULL) {
+          taosArrayDestroy(pResList);
+          taosArrayDestroy(pPageIdList);
           return terrno;
         }
 

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);
-  fprintf(stderr, "agg result: %f\n", *(double *)resultBuf.buf);
-
+  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/scalar/src/sclfunc.c

@@ -1178,9 +1178,7 @@ int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
   GET_TYPED_DATA(timeUnit, int64_t, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);
   GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
 
-  int64_t factor =
-      (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
-
+  int64_t factor = TSDB_TICK_PER_SECOND(timePrec);
   int64_t unit = timeUnit * 1000 / factor;
 
   for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
@@ -1372,9 +1370,7 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
     GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
   }
 
-  int64_t factor =
-      (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 : (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
-
+  int64_t factor = TSDB_TICK_PER_SECOND(timePrec);
   int32_t numOfRows = 0;
   for (int32_t i = 0; i < inputNum; ++i) {
     if (pInput[i].numOfRows > numOfRows) {

source/libs/scalar/src/sclvector.c

@@ -1426,57 +1426,6 @@ void vectorAssign(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pOut,
   pOut->numOfQualified = pRight->numOfQualified * pOut->numOfRows;
 }
 
-void vectorConcat(SScalarParam *pLeft, SScalarParam *pRight, void *out, int32_t _ord) {
-#if 0
-  int32_t len = pLeft->bytes + pRight->bytes;
-
-  int32_t i = ((_ord) == TSDB_ORDER_ASC) ? 0 : TMAX(pLeft->numOfRows, pRight->numOfRows) - 1;
-  int32_t step = ((_ord) == TSDB_ORDER_ASC) ? 1 : -1;
-
-  char *output = (char *)out;
-  if (pLeft->numOfRows == pRight->numOfRows) {
-    for (; i < pRight->numOfRows && i >= 0; i += step, output += len) {
-      char* left = POINTER_SHIFT(pLeft->data, pLeft->bytes * i);
-      char* right = POINTER_SHIFT(pRight->data, pRight->bytes * i);
-
-      if (isNull(left, pLeftCol->info.type) || isNull(right, pRight->info.type)) {
-        setVardataNull(output, TSDB_DATA_TYPE_BINARY);
-        continue;
-      }
-
-      // todo define a macro
-      memcpy(varDataVal(output), varDataVal(left), varDataLen(left));
-      memcpy(varDataVal(output) + varDataLen(left), varDataVal(right), varDataLen(right));
-      varDataSetLen(output, varDataLen(left) + varDataLen(right));
-    }
-  } else if (pLeft->numOfRows == 1) {
-    for (; i >= 0 && i < pRight->numOfRows; i += step, output += len) {
-      char *right = POINTER_SHIFT(pRight->data, pRight->bytes * i);
-      if (isNull(pLeft->data, pLeftCol->info.type) || isNull(right, pRight->info.type)) {
-        setVardataNull(output, TSDB_DATA_TYPE_BINARY);
-        continue;
-      }
-
-      memcpy(varDataVal(output), varDataVal(pLeft->data), varDataLen(pLeft->data));
-      memcpy(varDataVal(output) + varDataLen(pLeft->data), varDataVal(right), varDataLen(right));
-      varDataSetLen(output, varDataLen(pLeft->data) + varDataLen(right));
-    }
-  } else if (pRight->numOfRows == 1) {
-    for (; i >= 0 && i < pLeft->numOfRows; i += step, output += len) {
-      char* left = POINTER_SHIFT(pLeft->data, pLeft->bytes * i);
-      if (isNull(left, pLeftCol->info.type) || isNull(pRight->data, pRight->info.type)) {
-        SET_DOUBLE_NULL(output);
-        continue;
-      }
-
-      memcpy(varDataVal(output), varDataVal(left), varDataLen(pRight->data));
-      memcpy(varDataVal(output) + varDataLen(left), varDataVal(pRight->data), varDataLen(pRight->data));
-      varDataSetLen(output, varDataLen(left) + varDataLen(pRight->data));
-    }
-  }
-#endif
-}
-
 static void vectorBitAndHelper(SColumnInfoData *pLeftCol, SColumnInfoData *pRightCol, SColumnInfoData *pOutputCol,
                                int32_t numOfRows, int32_t step, int32_t i) {
   _getBigintValue_fn_t getVectorBigintValueFnLeft = getVectorBigintValueFn(pLeftCol->info.type);

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) {
-    memcpy(tmp, *pVal, valSize);
-    *pVal = tmp;
-    streamStateFreeCur(pCur);
-    return res;
-  }
-  streamStateFreeCur(pCur);
+  if (code == 0) {
+    if (key->win.skey <= tmpKey.win.skey && tmpKey.win.ekey <= key->win.ekey) {
+      memcpy(tmp, *pVal, valSize);
+      goto _end;
+    }
 
-  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/libs/tdb/src/db/tdbTable.c

@@ -53,6 +53,7 @@ int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprF
   if (strcmp(TDB_MAINDB_NAME, tbname)) {
     pPager = tdbEnvGetPager(pEnv, fFullName);
     if (!pPager) {
+      tdbOsFree(pTb);
       return -1;
     }
 
@@ -72,6 +73,7 @@ int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprF
     if (pPager == NULL) {
       ret = tdbPagerOpen(pEnv->pCache, fFullName, &pPager);
       if (ret < 0) {
+        tdbOsFree(pTb);
         return -1;
       }
 
@@ -94,6 +96,7 @@ int tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprF
     snprintf(fFullName, TDB_FILENAME_LEN, "%s/%s", pEnv->dbName, tbname);
     ret = tdbPagerOpen(pEnv->pCache, fFullName, &pPager);
     if (ret < 0) {
+      tdbOsFree(pTb);
       return -1;
     }
 

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;
-  cmd->screenOffset += taosWcharWidth(wc);
-  cmd->endOffset    += taosWcharWidth(wc);
+  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 };
 
@@ -143,7 +173,7 @@ typedef struct _script_t {
   char             *optionBuffer;
   SCmdLine         *lines;  // command list
   SVariable         variables[MAX_VAR_LEN];
-  TdThread         bgPid;
+  TdThread          bgPid;
   char              auth[128];
   struct _script_t *bgScripts[MAX_BACKGROUND_SCRIPT_NUM];
 } SScript;
@@ -182,4 +212,4 @@ bool     simExecuteLineInsertCmd(SScript *script, char *option);
 bool     simExecuteLineInsertErrorCmd(SScript *script, char *option);
 void     simVisuallizeOption(SScript *script, char *src, char *dst);
 
-#endif /*_TD_SIM_INT_H_*/
+#endif /*_TD_SIM_INT_H_*/
\ No newline at end of file

utils/tsim/src/simExe.c

@@ -657,11 +657,11 @@ bool simCreateTaosdConnect(SScript *script, char *rest) {
 }
 
 bool simExecuteNativeSqlCommand(SScript *script, char *rest, bool isSlow) {
-  char       timeStr[80] = {0};
-  time_t     tt;
-  struct tm  tp;
-  SCmdLine  *line = &script->lines[script->linePos];
-  int32_t    ret = -1;
+  char      timeStr[80] = {0};
+  time_t    tt;
+  struct tm tp;
+  SCmdLine *line = &script->lines[script->linePos];
+  int32_t   ret = -1;
 
   TAOS_RES *pSql = NULL;