Merge branch '3.0' into enh/stopquery

cmake/cmake.install

@@ -13,6 +13,7 @@ ELSEIF (TD_WINDOWS)
   INSTALL(FILES ${TD_SOURCE_DIR}/packaging/cfg/taos.cfg DESTINATION cfg)
   INSTALL(FILES ${TD_SOURCE_DIR}/include/client/taos.h DESTINATION include)
   INSTALL(FILES ${TD_SOURCE_DIR}/include/util/taoserror.h DESTINATION include)
+  INSTALL(FILES ${TD_SOURCE_DIR}/include/libs/function/taosudf.h DESTINATION include)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.lib DESTINATION driver)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos_static.lib DESTINATION driver)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.dll DESTINATION driver)

docs/examples/java/src/main/java/com/taos/example/LineProtocolExample.java

@@ -26,7 +26,7 @@ public class LineProtocolExample {
 
     private static void createDatabase(Connection conn) throws SQLException {
         try (Statement stmt = conn.createStatement()) {
-            // the default precision is ms (microsecond), but we use us(microsecond) here.
+            // the default precision is ms (millisecond), but we use us(microsecond) here.
             stmt.execute("CREATE DATABASE IF NOT EXISTS test PRECISION 'us'");
             stmt.execute("USE test");
         }

include/libs/function/taosudf.h

+/*
+* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+*
+* This program is free software: you can use, redistribute, and/or modify
+* it under the terms of the GNU Affero General Public License, version 3
+* or later ("AGPL"), as published by the Free Software Foundation.
+*
+* This program is distributed in the hope that it will be useful, but WITHOUT
+* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+* FITNESS FOR A PARTICULAR PURPOSE.
+*
+* You should have received a copy of the GNU Affero General Public License
+* along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef TDENGINE_TAOSUDF_H
+#define TDENGINE_TAOSUDF_H
+
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <taos.h>
+#include <taoserror.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#if defined(__GNUC__)
+#define FORCE_INLINE inline __attribute__((always_inline))
+#else
+#define FORCE_INLINE
+#endif
+typedef struct SUdfColumnMeta {
+  int16_t type;
+  int32_t bytes;
+  uint8_t precision;
+  uint8_t scale;
+} SUdfColumnMeta;
+
+typedef struct SUdfColumnData {
+  int32_t numOfRows;
+  int32_t rowsAlloc;
+  union {
+    struct {
+      int32_t nullBitmapLen;
+      char   *nullBitmap;
+      int32_t dataLen;
+      char   *data;
+    } fixLenCol;
+
+    struct {
+      int32_t varOffsetsLen;
+      int32_t   *varOffsets;
+      int32_t payloadLen;
+      char   *payload;
+      int32_t payloadAllocLen;
+    } varLenCol;
+  };
+} SUdfColumnData;
+
+
+typedef struct SUdfColumn {
+  SUdfColumnMeta colMeta;
+  bool           hasNull;
+  SUdfColumnData colData;
+} SUdfColumn;
+
+typedef struct SUdfDataBlock {
+  int32_t numOfRows;
+  int32_t numOfCols;
+  SUdfColumn **udfCols;
+} SUdfDataBlock;
+
+typedef struct SUdfInterBuf {
+  int32_t bufLen;
+  char* buf;
+  int8_t numOfResult; //zero or one
+} SUdfInterBuf;
+typedef void *UdfcFuncHandle;
+
+// dynamic lib init and destroy
+typedef int32_t (*TUdfInitFunc)();
+typedef int32_t (*TUdfDestroyFunc)();
+
+#define UDF_MEMORY_EXP_GROWTH 1.5
+#define NBIT                     (3u)
+#define BitPos(_n)               ((_n) & ((1 << NBIT) - 1))
+#define BMCharPos(bm_, r_)       ((bm_)[(r_) >> NBIT])
+#define BitmapLen(_n) (((_n) + ((1 << NBIT) - 1)) >> NBIT)
+
+#define udfColDataIsNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] == -1)
+#define udfColDataIsNull_f(pColumn, row) ((BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) & (1u << (7u - BitPos(row)))) == (1u << (7u - BitPos(row))))
+#define udfColDataSetNull_f(pColumn, row)                                                \
+  do {                                                                                   \
+    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) |= (1u << (7u - BitPos(row))); \
+  } while (0)
+
+#define udfColDataSetNotNull_f(pColumn, r_)                                               \
+  do {                                                                                    \
+    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, r_) &= ~(1u << (7u - BitPos(r_)));   \
+  } while (0)
+#define udfColDataSetNull_var(pColumn, row)  ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
+
+typedef uint16_t VarDataLenT;  // maxVarDataLen: 32767
+#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
+#define varDataLen(v) ((VarDataLenT *)(v))[0]
+#define varDataVal(v) ((char *)(v) + VARSTR_HEADER_SIZE)
+#define varDataTLen(v)         (sizeof(VarDataLenT) + varDataLen(v))
+#define varDataCopy(dst, v)    memcpy((dst), (void *)(v), varDataTLen(v))
+#define varDataLenByData(v)    (*(VarDataLenT *)(((char *)(v)) - VARSTR_HEADER_SIZE))
+#define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT)(_len))
+#define IS_VAR_DATA_TYPE(t) \
+  (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
+#define IS_STR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
+
+
+static FORCE_INLINE char* udfColDataGetData(const SUdfColumn* pColumn, int32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    return pColumn->colData.varLenCol.payload + pColumn->colData.varLenCol.varOffsets[row];
+  } else {
+    return pColumn->colData.fixLenCol.data + pColumn->colMeta.bytes * row;
+  }
+}
+
+static FORCE_INLINE bool udfColDataIsNull(const SUdfColumn* pColumn, int32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    if (pColumn->colMeta.type == TSDB_DATA_TYPE_JSON) {
+      if (udfColDataIsNull_var(pColumn, row)) {
+        return true;
+      }
+      char* data = udfColDataGetData(pColumn, row);
+      return (*data == TSDB_DATA_TYPE_NULL);
+    } else {
+      return udfColDataIsNull_var(pColumn, row);
+    }
+  } else {
+    return udfColDataIsNull_f(pColumn, row);
+  }
+}
+
+static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t newCapacity) {
+  SUdfColumnMeta *meta = &pColumn->colMeta;
+  SUdfColumnData *data = &pColumn->colData;
+
+  if (newCapacity== 0 || newCapacity <= data->rowsAlloc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int allocCapacity = (data->rowsAlloc< 8) ? 8 : data->rowsAlloc;
+  while (allocCapacity < newCapacity) {
+    allocCapacity *= UDF_MEMORY_EXP_GROWTH;
+  }
+
+  if (IS_VAR_DATA_TYPE(meta->type)) {
+    char* tmp = (char*)realloc(data->varLenCol.varOffsets, sizeof(int32_t) * allocCapacity);
+    if (tmp == NULL) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    data->varLenCol.varOffsets = (int32_t*)tmp;
+    data->varLenCol.varOffsetsLen = sizeof(int32_t) * allocCapacity;
+    // for payload, add data in udfColDataAppend
+  } else {
+    char* tmp = (char*)realloc(data->fixLenCol.nullBitmap, BitmapLen(allocCapacity));
+    if (tmp == NULL) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    data->fixLenCol.nullBitmap = tmp;
+    data->fixLenCol.nullBitmapLen = BitmapLen(allocCapacity);
+    if (meta->type == TSDB_DATA_TYPE_NULL) {
+      return TSDB_CODE_SUCCESS;
+    }
+
+    tmp = (char*)realloc(data->fixLenCol.data, allocCapacity* meta->bytes);
+    if (tmp == NULL) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+
+    data->fixLenCol.data = tmp;
+    data->fixLenCol.dataLen = allocCapacity* meta->bytes;
+  }
+
+  data->rowsAlloc = allocCapacity;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static FORCE_INLINE void udfColDataSetNull(SUdfColumn* pColumn, int32_t row) {
+  udfColEnsureCapacity(pColumn, row+1);
+  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
+    udfColDataSetNull_var(pColumn, row);
+  } else {
+    udfColDataSetNull_f(pColumn, row);
+  }
+  pColumn->hasNull = true;
+}
+
+static FORCE_INLINE int32_t udfColDataSet(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {
+  SUdfColumnMeta *meta = &pColumn->colMeta;
+  SUdfColumnData *data = &pColumn->colData;
+  udfColEnsureCapacity(pColumn, currentRow+1);
+  bool isVarCol = IS_VAR_DATA_TYPE(meta->type);
+  if (isNull) {
+    udfColDataSetNull(pColumn, currentRow);
+  } else {
+    if (!isVarCol) {
+      udfColDataSetNotNull_f(pColumn, currentRow);
+      memcpy(data->fixLenCol.data + meta->bytes * currentRow, pData, meta->bytes);
+    } else {
+      int32_t dataLen = varDataTLen(pData);
+      if (meta->type == TSDB_DATA_TYPE_JSON) {
+        if (*pData == TSDB_DATA_TYPE_NULL) {
+          dataLen = 0;
+        } else if (*pData == TSDB_DATA_TYPE_NCHAR) {
+          dataLen = varDataTLen(pData + sizeof(char));
+        } else if (*pData == TSDB_DATA_TYPE_BIGINT || *pData == TSDB_DATA_TYPE_DOUBLE) {
+          dataLen = sizeof(int64_t);
+        } else if (*pData == TSDB_DATA_TYPE_BOOL) {
+          dataLen = sizeof(char);
+        }
+        dataLen += sizeof(char);
+      }
+
+      if (data->varLenCol.payloadAllocLen < data->varLenCol.payloadLen + dataLen) {
+        uint32_t newSize = data->varLenCol.payloadAllocLen;
+        if (newSize <= 1) {
+          newSize = 8;
+        }
+
+        while (newSize < data->varLenCol.payloadLen + dataLen) {
+          newSize = newSize * UDF_MEMORY_EXP_GROWTH;
+        }
+
+        char *buf = (char*)realloc(data->varLenCol.payload, newSize);
+        if (buf == NULL) {
+          return TSDB_CODE_OUT_OF_MEMORY;
+        }
+
+        data->varLenCol.payload = buf;
+        data->varLenCol.payloadAllocLen = newSize;
+      }
+
+      uint32_t len = data->varLenCol.payloadLen;
+      data->varLenCol.varOffsets[currentRow] = len;
+
+      memcpy(data->varLenCol.payload + len, pData, dataLen);
+      data->varLenCol.payloadLen += dataLen;
+    }
+  }
+  data->numOfRows = (currentRow + 1 > data->numOfRows) ? (currentRow+1) : data->numOfRows;
+  return 0;
+}
+
+typedef int32_t (*TUdfScalarProcFunc)(SUdfDataBlock* block, SUdfColumn *resultCol);
+
+typedef int32_t (*TUdfAggStartFunc)(SUdfInterBuf *buf);
+typedef int32_t (*TUdfAggProcessFunc)(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf);
+typedef int32_t (*TUdfAggFinishFunc)(SUdfInterBuf* buf, SUdfInterBuf *resultData);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_TAOSUDF_H

include/libs/function/tudf.h

@@ -16,6 +16,13 @@
 #ifndef TDENGINE_TUDF_H
 #define TDENGINE_TUDF_H
 
+#undef malloc
+#define malloc malloc
+#undef free
+#define free free
+#undef realloc
+#define alloc alloc
+#include <taosudf.h>
 
 #include <stdint.h>
 #include <stdbool.h>
@@ -36,56 +43,6 @@ extern "C" {
 #endif
 #define UDF_DNODE_ID_ENV_NAME "DNODE_ID"
 
-//======================================================================================
-//begin API to taosd and qworker
-
-typedef struct SUdfColumnMeta {
-  int16_t type;
-  int32_t bytes;
-  uint8_t precision;
-  uint8_t scale;
-} SUdfColumnMeta;
-
-typedef struct SUdfColumnData {
-  int32_t numOfRows;
-  int32_t rowsAlloc;
-  union {
-    struct {
-      int32_t nullBitmapLen;
-      char   *nullBitmap;
-      int32_t dataLen;
-      char   *data;
-    } fixLenCol;
-
-    struct {
-      int32_t varOffsetsLen;
-      int32_t   *varOffsets;
-      int32_t payloadLen;
-      char   *payload;
-      int32_t payloadAllocLen;
-    } varLenCol;
-  };
-} SUdfColumnData;
-
-
-typedef struct SUdfColumn {
-  SUdfColumnMeta colMeta;
-  bool           hasNull;
-  SUdfColumnData colData;
-} SUdfColumn;
-
-typedef struct SUdfDataBlock {
-  int32_t numOfRows;
-  int32_t numOfCols;
-  SUdfColumn **udfCols;
-} SUdfDataBlock;
-
-typedef struct SUdfInterBuf {
-  int32_t bufLen;
-  char* buf;
-  int8_t numOfResult; //zero or one
-} SUdfInterBuf;
-typedef void *UdfcFuncHandle;
 
 //low level APIs
 /**
@@ -127,177 +84,6 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock);
 int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols, SScalarParam *output);
 
 int32_t cleanUpUdfs();
-// end API to taosd and qworker
-//=============================================================================================================================
-// begin API to UDF writer.
-
-// dynamic lib init and destroy
-typedef int32_t (*TUdfInitFunc)();
-typedef int32_t (*TUdfDestroyFunc)();
-
-//TODO: add API to check function arguments type, number etc.
-
-#define UDF_MEMORY_EXP_GROWTH 1.5
-
-#define udfColDataIsNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] == -1)
-#define udfColDataIsNull_f(pColumn, row) ((BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) & (1u << (7u - BitPos(row)))) == (1u << (7u - BitPos(row))))
-#define udfColDataSetNull_f(pColumn, row)                                                \
-  do {                                                                                   \
-    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) |= (1u << (7u - BitPos(row))); \
-  } while (0)
-
-#define udfColDataSetNotNull_f(pColumn, r_)                                               \
-  do {                                                                                    \
-    BMCharPos(pColumn->colData.fixLenCol.nullBitmap, r_) &= ~(1u << (7u - BitPos(r_)));   \
-  } while (0)
-#define udfColDataSetNull_var(pColumn, row)  ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
-
-
-static FORCE_INLINE char* udfColDataGetData(const SUdfColumn* pColumn, int32_t row) {
-  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
-    return pColumn->colData.varLenCol.payload + pColumn->colData.varLenCol.varOffsets[row];
-  } else {
-    return pColumn->colData.fixLenCol.data + pColumn->colMeta.bytes * row;
-  }
-}
-
-static FORCE_INLINE bool udfColDataIsNull(const SUdfColumn* pColumn, int32_t row) {
-  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
-    if (pColumn->colMeta.type == TSDB_DATA_TYPE_JSON) {
-      if (udfColDataIsNull_var(pColumn, row)) {
-        return true;
-      }
-      char* data = udfColDataGetData(pColumn, row);
-      return (*data == TSDB_DATA_TYPE_NULL);
-    } else {
-      return udfColDataIsNull_var(pColumn, row);
-    }
-  } else {
-    return udfColDataIsNull_f(pColumn, row);
-  }
-}
-
-static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t newCapacity) {
-  SUdfColumnMeta *meta = &pColumn->colMeta;
-  SUdfColumnData *data = &pColumn->colData;
-
-  if (newCapacity== 0 || newCapacity <= data->rowsAlloc) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  int allocCapacity = TMAX(data->rowsAlloc, 8);
-  while (allocCapacity < newCapacity) {
-    allocCapacity *= UDF_MEMORY_EXP_GROWTH;
-  }
-
-  if (IS_VAR_DATA_TYPE(meta->type)) {
-    char* tmp = taosMemoryRealloc(data->varLenCol.varOffsets, sizeof(int32_t) * allocCapacity);
-    if (tmp == NULL) {
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-    data->varLenCol.varOffsets = (int32_t*)tmp;
-    data->varLenCol.varOffsetsLen = sizeof(int32_t) * allocCapacity;
-    // for payload, add data in udfColDataAppend
-  } else {
-    char* tmp = taosMemoryRealloc(data->fixLenCol.nullBitmap, BitmapLen(allocCapacity));
-    if (tmp == NULL) {
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-    data->fixLenCol.nullBitmap = tmp;
-    data->fixLenCol.nullBitmapLen = BitmapLen(allocCapacity);
-    if (meta->type == TSDB_DATA_TYPE_NULL) {
-      return TSDB_CODE_SUCCESS;
-    }
-
-    tmp = taosMemoryRealloc(data->fixLenCol.data, allocCapacity* meta->bytes);
-    if (tmp == NULL) {
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-
-    data->fixLenCol.data = tmp;
-    data->fixLenCol.dataLen = allocCapacity* meta->bytes;
-  }
-
-  data->rowsAlloc = allocCapacity;
-
-  return TSDB_CODE_SUCCESS;
-}
-
-static FORCE_INLINE void udfColDataSetNull(SUdfColumn* pColumn, int32_t row) {
-  udfColEnsureCapacity(pColumn, row+1);
-  if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
-    udfColDataSetNull_var(pColumn, row);
-  } else {
-    udfColDataSetNull_f(pColumn, row);
-  }
-  pColumn->hasNull = true;
-}
-
-static FORCE_INLINE int32_t udfColDataSet(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {
-  SUdfColumnMeta *meta = &pColumn->colMeta;
-  SUdfColumnData *data = &pColumn->colData;
-  udfColEnsureCapacity(pColumn, currentRow+1);
-  bool isVarCol = IS_VAR_DATA_TYPE(meta->type);
-  if (isNull) {
-      udfColDataSetNull(pColumn, currentRow);
-  } else {
-    if (!isVarCol) {
-      colDataSetNotNull_f(data->fixLenCol.nullBitmap, currentRow);
-      memcpy(data->fixLenCol.data + meta->bytes * currentRow, pData, meta->bytes);
-    } else {
-      int32_t dataLen = varDataTLen(pData);
-      if (meta->type == TSDB_DATA_TYPE_JSON) {
-        if (*pData == TSDB_DATA_TYPE_NULL) {
-          dataLen = 0;
-        } else if (*pData == TSDB_DATA_TYPE_NCHAR) {
-          dataLen = varDataTLen(pData + CHAR_BYTES);
-        } else if (*pData == TSDB_DATA_TYPE_BIGINT || *pData == TSDB_DATA_TYPE_DOUBLE) {
-          dataLen = LONG_BYTES;
-        } else if (*pData == TSDB_DATA_TYPE_BOOL) {
-          dataLen = CHAR_BYTES;
-        }
-        dataLen += CHAR_BYTES;
-      }
-
-      if (data->varLenCol.payloadAllocLen < data->varLenCol.payloadLen + dataLen) {
-        uint32_t newSize = data->varLenCol.payloadAllocLen;
-        if (newSize <= 1) {
-          newSize = 8;
-        }
-
-        while (newSize < data->varLenCol.payloadLen + dataLen) {
-          newSize = newSize * UDF_MEMORY_EXP_GROWTH;
-        }
-
-        char *buf = taosMemoryRealloc(data->varLenCol.payload, newSize);
-        if (buf == NULL) {
-          return TSDB_CODE_OUT_OF_MEMORY;
-        }
-
-        data->varLenCol.payload = buf;
-        data->varLenCol.payloadAllocLen = newSize;
-      }
-
-      uint32_t len = data->varLenCol.payloadLen;
-      data->varLenCol.varOffsets[currentRow] = len;
-
-      memcpy(data->varLenCol.payload + len, pData, dataLen);
-      data->varLenCol.payloadLen += dataLen;
-    }
-  }
-  data->numOfRows = TMAX(currentRow + 1, data->numOfRows);
-  return 0;
-}
-
-typedef int32_t (*TUdfScalarProcFunc)(SUdfDataBlock* block, SUdfColumn *resultCol);
-
-typedef int32_t (*TUdfAggStartFunc)(SUdfInterBuf *buf);
-typedef int32_t (*TUdfAggProcessFunc)(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf);
-typedef int32_t (*TUdfAggFinishFunc)(SUdfInterBuf* buf, SUdfInterBuf *resultData);
-
-
-// end API to UDF writer
-//=======================================================================================================================
 
 #ifdef __cplusplus
 }

include/libs/nodes/plannodes.h

@@ -103,6 +103,7 @@ typedef struct SIndefRowsFuncLogicNode {
   SLogicNode node;
   SNodeList* pFuncs;
   bool       isTailFunc;
+  bool       isUniqueFunc;
 } SIndefRowsFuncLogicNode;
 
 typedef struct SInterpFuncLogicNode {

include/libs/nodes/querynodes.h

@@ -50,6 +50,7 @@ typedef struct SExprNode {
   char      aliasName[TSDB_COL_NAME_LEN];
   char      userAlias[TSDB_COL_NAME_LEN];
   SArray*   pAssociation;
+  bool      orderAlias;
 } SExprNode;
 
 typedef enum EColumnType { COLUMN_TYPE_COLUMN = 1, COLUMN_TYPE_TAG, COLUMN_TYPE_TBNAME } EColumnType;

include/libs/planner/planner.h

@@ -48,6 +48,8 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
 // @pSource one execution location of this group of datasource subplans
 int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
 
+int32_t qClearSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId);
+
 // Convert to subplan to string for the scheduler to send to the executor
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);
 int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan);

include/libs/qcom/query.h

@@ -271,19 +271,19 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define qDebug(...)                                                                     \
   do {                                                                                  \
     if (qDebugFlag & DEBUG_DEBUG) {                                                     \
-      taosPrintLog("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__);                       \
+      taosPrintLog("QRY ", DEBUG_DEBUG, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
     }                                                                                   \
   } while (0)
 #define qTrace(...)                                                                     \
   do {                                                                                  \
     if (qDebugFlag & DEBUG_TRACE) {                                                     \
-      taosPrintLog("QRY ", DEBUG_TRACE, qDebugFlag, __VA_ARGS__);                       \
+      taosPrintLog("QRY ", DEBUG_TRACE, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
     }                                                                                   \
   } while (0)
 #define qDebugL(...)                                                                           \
   do {                                                                                         \
     if (qDebugFlag & DEBUG_DEBUG) {                                                            \
-      taosPrintLongString("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__);                       \
+      taosPrintLongString("QRY ", DEBUG_DEBUG, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
     }                                                                                          \
   } while (0)
 

include/util/taoserror.h

@@ -25,7 +25,7 @@ extern "C" {
 // clang-format off
 
 #define TAOS_DEF_ERROR_CODE(mod, code) ((int32_t)((0x80000000 | ((mod)<<16) | (code))))
- 
+
 #define TAOS_SYSTEM_ERROR(code)             (0x80ff0000 | (code))
 #define TAOS_SUCCEEDED(err)                 ((err) >= 0)
 #define TAOS_FAILED(err)                    ((err) < 0)
@@ -35,7 +35,7 @@ const char* terrstr();
 
 int32_t* taosGetErrno();
 #define terrno                              (*taosGetErrno())
- 
+
 #define TSDB_CODE_SUCCESS                   0
 #define TSDB_CODE_FAILED                    -1   // unknown or needn't tell detail error
 
@@ -615,6 +615,7 @@ int32_t* taosGetErrno();
 //rsma
 #define TSDB_CODE_RSMA_INVALID_ENV               TAOS_DEF_ERROR_CODE(0, 0x3150)
 #define TSDB_CODE_RSMA_INVALID_STAT              TAOS_DEF_ERROR_CODE(0, 0x3151)
+#define TSDB_CODE_RSMA_QTASKINFO_CREATE          TAOS_DEF_ERROR_CODE(0, 0x3152)
 
 //index
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)

packaging/check_package.sh

@@ -170,7 +170,7 @@ function check_lib_path() {
 
 function check_header_path() {
 	# check all header
-	header_dir=("taos.h" "taosdef.h" "taoserror.h")
+	header_dir=("taos.h" "taosdef.h" "taoserror.h" "taosudf.h")
     for i in "${header_dir[@]}";do
         check_link ${inc_link_dir}/$i
     done

packaging/deb/DEBIAN/prerm

@@ -29,6 +29,7 @@ else
     ${csudo}rm -f ${bin_link_dir}/taosdemo   || :
     ${csudo}rm -f ${cfg_link_dir}/*          || :
     ${csudo}rm -f ${inc_link_dir}/taos.h     || :
+    ${csudo}rm -f ${inc_link_dir}/taosudf.h     || :
     ${csudo}rm -f ${lib_link_dir}/libtaos.*  || :
 
     ${csudo}rm -f ${log_link_dir}            || :

packaging/deb/makedeb.sh

@@ -70,6 +70,7 @@ cp ${compile_dir}/build/lib/${libfile}              ${pkg_dir}${install_home_pat
 cp ${compile_dir}/../include/client/taos.h          ${pkg_dir}${install_home_path}/include
 cp ${compile_dir}/../include/common/taosdef.h       ${pkg_dir}${install_home_path}/include
 cp ${compile_dir}/../include/util/taoserror.h       ${pkg_dir}${install_home_path}/include
+cp ${compile_dir}/../include/libs/function/taosudf.h       ${pkg_dir}${install_home_path}/include
 cp -r ${top_dir}/examples/*                         ${pkg_dir}${install_home_path}/examples
 #cp -r ${top_dir}/src/connector/python               ${pkg_dir}${install_home_path}/connector
 #cp -r ${top_dir}/src/connector/go                   ${pkg_dir}${install_home_path}/connector

packaging/rpm/tdengine.spec

@@ -77,6 +77,7 @@ cp %{_compiledir}/build/lib/${libfile}              %{buildroot}%{homepath}/driv
 cp %{_compiledir}/../include/client/taos.h          %{buildroot}%{homepath}/include
 cp %{_compiledir}/../include/common/taosdef.h       %{buildroot}%{homepath}/include
 cp %{_compiledir}/../include/util/taoserror.h       %{buildroot}%{homepath}/include
+cp %{_compiledir}/../include/libs/function/taosudf.h       %{buildroot}%{homepath}/include
 #cp -r %{_compiledir}/../src/connector/python        %{buildroot}%{homepath}/connector
 #cp -r %{_compiledir}/../src/connector/go            %{buildroot}%{homepath}/connector
 #cp -r %{_compiledir}/../src/connector/nodejs        %{buildroot}%{homepath}/connector
@@ -201,6 +202,7 @@ if [ $1 -eq 0 ];then
     ${csudo}rm -f ${inc_link_dir}/taos.h     || :
     ${csudo}rm -f ${inc_link_dir}/taosdef.h     || :
     ${csudo}rm -f ${inc_link_dir}/taoserror.h     || :
+    ${csudo}rm -f ${inc_link_dir}/taosudf.h     || :    
     ${csudo}rm -f ${lib_link_dir}/libtaos.*  || :
 
     ${csudo}rm -f ${log_link_dir}            || :

packaging/tools/install.sh

@@ -314,11 +314,12 @@ function install_jemalloc() {
 }
 
 function install_header() {
-  ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h || :
+  ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h ${inc_link_dir}/taosudf.h || :
   ${csudo}cp -f ${script_dir}/inc/* ${install_main_dir}/include && ${csudo}chmod 644 ${install_main_dir}/include/*
   ${csudo}ln -s ${install_main_dir}/include/taos.h ${inc_link_dir}/taos.h
   ${csudo}ln -s ${install_main_dir}/include/taosdef.h ${inc_link_dir}/taosdef.h
   ${csudo}ln -s ${install_main_dir}/include/taoserror.h ${inc_link_dir}/taoserror.h
+  ${csudo}ln -s ${install_main_dir}/include/taosudf.h ${inc_link_dir}/taosudf.h  
 }
 
 function add_newHostname_to_hosts() {

packaging/tools/install_arbi.sh

@@ -115,11 +115,12 @@ function install_bin() {
 }
 
 function install_header() {
-  ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h || :
+  ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h ${inc_link_dir}/taosudf.h || :
   ${csudo}cp -f ${script_dir}/inc/* ${install_main_dir}/include && ${csudo}chmod 644 ${install_main_dir}/include/*
   ${csudo}ln -s ${install_main_dir}/include/taos.h ${inc_link_dir}/taos.h
   ${csudo}ln -s ${install_main_dir}/include/taosdef.h ${inc_link_dir}/taosdef.h
   ${csudo}ln -s ${install_main_dir}/include/taoserror.h ${inc_link_dir}/taoserror.h
+  ${csudo}ln -s ${install_main_dir}/include/taosudf.h ${inc_link_dir}/taosudf.h
 }
 
 function install_jemalloc() {

packaging/tools/install_client.sh

@@ -148,11 +148,12 @@ function install_lib() {
 }
 
 function install_header() {
-    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h    || :
+    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h  ${inc_link_dir}/taosudf.h  || :
     ${csudo}cp -f ${script_dir}/inc/* ${install_main_dir}/include && ${csudo}chmod 644 ${install_main_dir}/include/*
     ${csudo}ln -s ${install_main_dir}/include/taos.h ${inc_link_dir}/taos.h
     ${csudo}ln -s ${install_main_dir}/include/taosdef.h ${inc_link_dir}/taosdef.h
     ${csudo}ln -s ${install_main_dir}/include/taoserror.h ${inc_link_dir}/taoserror.h
+    ${csudo}ln -s ${install_main_dir}/include/taosudf.h ${inc_link_dir}/taosudf.h    
 }
 
 function install_jemalloc() {

packaging/tools/make_install.sh

@@ -349,16 +349,17 @@ function install_lib() {
 function install_header() {
 
   if [ "$osType" != "Darwin" ]; then
-    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h || :
-    ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h \
+    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h ${inc_link_dir}/taosudf.h || :
+    ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h ${source_dir}/include/libs/function/taosudf.h \
       ${install_main_dir}/include && ${csudo}chmod 644 ${install_main_dir}/include/*
     ${csudo}ln -s ${install_main_dir}/include/taos.h ${inc_link_dir}/taos.h
     ${csudo}ln -s ${install_main_dir}/include/taosdef.h ${inc_link_dir}/taosdef.h
     ${csudo}ln -s ${install_main_dir}/include/taoserror.h ${inc_link_dir}/taoserror.h
+    ${csudo}ln -s ${install_main_dir}/include/taosudf.h ${inc_link_dir}/taosudf.h
   else
-    ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h \
+    ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h ${source_dir}/include/libs/function/taosudf.h \
       ${install_main_dir}/include ||
-      ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h \
+      ${csudo}cp -f ${source_dir}/include/client/taos.h ${source_dir}/include/common/taosdef.h ${source_dir}/include/util/taoserror.h ${source_dir}/include/libs/function/taosudf.h \
         ${install_main_2_dir}/include &&
       ${csudo}chmod 644 ${install_main_dir}/include/* ||
       ${csudo}chmod 644 ${install_main_2_dir}/include/*

packaging/tools/makearbi.sh

@@ -36,7 +36,7 @@ fi
 bin_files="${build_dir}/bin/tarbitrator ${script_dir}/remove_arbi.sh"
 install_files="${script_dir}/install_arbi.sh"
 
-#header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
+#header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h ${code_dir}/include/libs/function/taosudf.h"
 init_file_tarbitrator_deb=${script_dir}/../deb/tarbitratord
 init_file_tarbitrator_rpm=${script_dir}/../rpm/tarbitratord
 

packaging/tools/makeclient.sh

@@ -62,7 +62,7 @@ else
   lib_files="${build_dir}/lib/libtaos.${version}.dylib"
 fi
 
-header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
+header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h ${code_dir}/include/libs/function/taosudf.h"
 if [ "$dbName" != "taos" ]; then
   cfg_dir="${top_dir}/../enterprise/packaging/cfg"
 else

packaging/tools/makepkg.sh

@@ -93,7 +93,7 @@ else
 fi
 
 lib_files="${build_dir}/lib/libtaos.so.${version}"
-header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
+header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h ${code_dir}/include/libs/function/taosudf.h"
 
 if [ "$dbName" != "taos" ]; then
   cfg_dir="${top_dir}/../enterprise/packaging/cfg"

packaging/tools/post.sh

@@ -81,10 +81,11 @@ function kill_taosd() {
 }
 
 function install_include() {
-    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h|| :
+    ${csudo}rm -f ${inc_link_dir}/taos.h ${inc_link_dir}/taosdef.h ${inc_link_dir}/taoserror.h ${inc_link_dir}/taosudf.h || :
     ${csudo}ln -s ${inc_dir}/taos.h ${inc_link_dir}/taos.h
     ${csudo}ln -s ${inc_dir}/taosdef.h ${inc_link_dir}/taosdef.h
     ${csudo}ln -s ${inc_dir}/taoserror.h ${inc_link_dir}/taoserror.h
+    ${csudo}ln -s ${inc_dir}/taosudf.h ${inc_link_dir}/taosudf.h    
 }
 
 function install_lib() {

packaging/tools/preun.sh

@@ -128,6 +128,7 @@ ${csudo}rm -f ${cfg_link_dir}/*.new      || :
 ${csudo}rm -f ${inc_link_dir}/taos.h     || :
 ${csudo}rm -f ${inc_link_dir}/taosdef.h  || :
 ${csudo}rm -f ${inc_link_dir}/taoserror.h || :
+${csudo}rm -f ${inc_link_dir}/taosudf.h || :
 ${csudo}rm -f ${lib_link_dir}/libtaos.*   || :
 ${csudo}rm -f ${lib64_link_dir}/libtaos.* || :
 

packaging/tools/remove.sh

@@ -84,6 +84,7 @@ function clean_header() {
     ${csudo} rm -f ${inc_link_dir}/taos.h       || :
     ${csudo} rm -f ${inc_link_dir}/taosdef.h    || :
     ${csudo} rm -f ${inc_link_dir}/taoserror.h  || :
+    ${csudo} rm -f ${inc_link_dir}/taosudf.h  || :
 }
 
 function clean_config() {

packaging/tools/remove_arbi.sh

@@ -59,6 +59,8 @@ function clean_header() {
     ${csudo}rm -f ${inc_link_dir}/taos.h       || :
     ${csudo}rm -f ${inc_link_dir}/taosdef.h    || :
     ${csudo}rm -f ${inc_link_dir}/taoserror.h  || :
+    ${csudo}rm -f ${inc_link_dir}/taosudf.h  || :
+
 }
 
 function clean_log() {

packaging/tools/remove_client.sh

@@ -54,6 +54,7 @@ function clean_header() {
     ${csudo}rm -f ${inc_link_dir}/taos.h           || :
     ${csudo}rm -f ${inc_link_dir}/taosdef.h        || :
     ${csudo}rm -f ${inc_link_dir}/taoserror.h      || :
+    ${csudo}rm -f ${inc_link_dir}/taosudf.h      || :    
 }
 
 function clean_config() {

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

@@ -15,7 +15,6 @@
 
 #define _DEFAULT_SOURCE
 #include "mndAcct.h"
-#include "mndPrivilege.h"
 #include "mndBnode.h"
 #include "mndCluster.h"
 #include "mndConsumer.h"
@@ -27,6 +26,7 @@
 #include "mndMnode.h"
 #include "mndOffset.h"
 #include "mndPerfSchema.h"
+#include "mndPrivilege.h"
 #include "mndProfile.h"
 #include "mndQnode.h"
 #include "mndQuery.h"
@@ -416,7 +416,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
     char          *syncNodeStr = sync2SimpleStr(pMgmt->sync);
     static int64_t mndTick = 0;
     if (++mndTick % 10 == 1) {
-      mTrace("vgId:%d, sync heartbeat msg:%s, %s", syncGetVgId(pMgmt->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
+      mTrace("vgId:%d, sync trace msg:%s, %s", syncGetVgId(pMgmt->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
     }
     if (gRaftDetailLog) {
       char logBuf[512] = {0};

source/dnode/vnode/src/inc/sma.h

@@ -205,16 +205,16 @@ struct STFile {
   uint8_t   state;
 };
 
-#define TD_FILE_F(tf)            (&((tf)->f))
-#define TD_FILE_PFILE(tf)        ((tf)->pFile)
-#define TD_FILE_OPENED(tf)       (TD_FILE_PFILE(tf) != NULL)
-#define TD_FILE_FULL_NAME(tf)    (TD_FILE_F(tf)->aname)
-#define TD_FILE_REL_NAME(tf)     (TD_FILE_F(tf)->rname)
-#define TD_FILE_OPENED(tf)       (TD_FILE_PFILE(tf) != NULL)
-#define TD_FILE_CLOSED(tf)       (!TD_FILE_OPENED(tf))
-#define TD_FILE_SET_CLOSED(f)    (TD_FILE_PFILE(f) = NULL)
-#define TD_FILE_SET_STATE(tf, s) ((tf)->state = (s))
-#define TD_FILE_DID(tf)          (TD_FILE_F(tf)->did)
+#define TD_TFILE_F(tf)            (&((tf)->f))
+#define TD_TFILE_PFILE(tf)        ((tf)->pFile)
+#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
+#define TD_TFILE_FULL_NAME(tf)    (TD_TFILE_F(tf)->aname)
+#define TD_TFILE_REL_NAME(tf)     (TD_TFILE_F(tf)->rname)
+#define TD_TFILE_OPENED(tf)       (TD_TFILE_PFILE(tf) != NULL)
+#define TD_TFILE_CLOSED(tf)       (!TD_TFILE_OPENED(tf))
+#define TD_TFILE_SET_CLOSED(f)    (TD_TFILE_PFILE(f) = NULL)
+#define TD_TFILE_SET_STATE(tf, s) ((tf)->state = (s))
+#define TD_TFILE_DID(tf)          (TD_TFILE_F(tf)->did)
 
 int32_t tdInitTFile(STFile *pTFile, STfs *pTfs, const char *fname);
 int32_t tdCreateTFile(STFile *pTFile, STfs *pTfs, bool updateHeader, int8_t fType);

source/dnode/vnode/src/inc/vnodeInt.h

@@ -64,6 +64,7 @@ typedef struct STsdbSnapshotReader STsdbSnapshotReader;
 #define VNODE_TQ_DIR    "tq"
 #define VNODE_WAL_DIR   "wal"
 #define VNODE_TSMA_DIR  "tsma"
+#define VNODE_RSMA_DIR  "rsma"
 #define VNODE_RSMA0_DIR "tsdb"
 #define VNODE_RSMA1_DIR "rsma1"
 #define VNODE_RSMA2_DIR "rsma2"
@@ -161,7 +162,6 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool
 
 // sma
 int32_t smaOpen(SVnode* pVnode);
-int32_t smaClose(SSma* pSma);
 int32_t smaCloseEnv(SSma* pSma);
 int32_t smaCloseEx(SSma* pSma);
 

source/dnode/vnode/src/sma/smaOpen.c

@@ -123,7 +123,7 @@ int32_t smaOpen(SVnode *pVnode) {
     }
 
     // restore the rsma
-#if 0
+#if 1
     if (rsmaRestore(pSma) < 0) {
       goto _err;
     }
@@ -154,12 +154,6 @@ int32_t smaCloseEx(SSma *pSma) {
   return 0;
 }
 
-int32_t smaClose(SSma *pSma) {
-  smaCloseEnv(pSma);
-  smaCloseEx(pSma);
-  return 0;
-}
-
 /**
  * @brief rsma env restore
  *

source/dnode/vnode/src/sma/smaUtil.c

@@ -22,7 +22,6 @@
 
 #define TD_FILE_INIT_MAGIC 0xFFFFFFFF
 
-
 static int32_t tdEncodeTFInfo(void **buf, STFInfo *pInfo);
 static void   *tdDecodeTFInfo(void *buf, STFInfo *pInfo);
 
@@ -46,7 +45,7 @@ static void *tdDecodeTFInfo(void *buf, STFInfo *pInfo) {
 }
 
 int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte) {
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
 
   int64_t nwrite = taosWriteFile(pTFile->pFile, buf, nbyte);
   if (nwrite < nbyte) {
@@ -58,9 +57,9 @@ int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte) {
 }
 
 int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence) {
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
 
-  int64_t loffset = taosLSeekFile(TD_FILE_PFILE(pTFile), offset, whence);
+  int64_t loffset = taosLSeekFile(TD_TFILE_PFILE(pTFile), offset, whence);
   if (loffset < 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
@@ -70,12 +69,12 @@ int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence) {
 }
 
 int64_t tdGetTFileSize(STFile *pTFile, int64_t *size) {
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
   return taosFStatFile(pTFile->pFile, size, NULL);
 }
 
 int64_t tdReadTFile(STFile *pTFile, void *buf, int64_t nbyte) {
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
 
   int64_t nread = taosReadFile(pTFile->pFile, buf, nbyte);
   if (nread < 0) {
@@ -108,7 +107,7 @@ int32_t tdLoadTFileHeader(STFile *pTFile, STFInfo *pInfo) {
   char     buf[TD_FILE_HEAD_SIZE] = "\0";
   uint32_t _version;
 
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
 
   if (tdSeekTFile(pTFile, 0, SEEK_SET) < 0) {
     return -1;
@@ -133,7 +132,7 @@ void tdUpdateTFileMagic(STFile *pTFile, void *pCksm) {
 }
 
 int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset) {
-  ASSERT(TD_FILE_OPENED(pTFile));
+  ASSERT(TD_TFILE_OPENED(pTFile));
 
   int64_t toffset;
 
@@ -141,6 +140,11 @@ int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset)
     return -1;
   }
 
+#if 1
+  smaDebug("append to file %s, offset:%" PRIi64 " + nbyte:%" PRIi64 " =%" PRIi64, TD_TFILE_FULL_NAME(pTFile), toffset,
+           nbyte, toffset + nbyte);
+#endif
+
   ASSERT(pTFile->info.fsize == toffset);
 
   if (offset) {
@@ -157,9 +161,9 @@ int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset)
 }
 
 int32_t tdOpenTFile(STFile *pTFile, int flags) {
-  ASSERT(!TD_FILE_OPENED(pTFile));
+  ASSERT(!TD_TFILE_OPENED(pTFile));
 
-  pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), flags);
+  pTFile->pFile = taosOpenFile(TD_TFILE_FULL_NAME(pTFile), flags);
   if (pTFile->pFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
@@ -169,9 +173,9 @@ int32_t tdOpenTFile(STFile *pTFile, int flags) {
 }
 
 void tdCloseTFile(STFile *pTFile) {
-  if (TD_FILE_OPENED(pTFile)) {
+  if (TD_TFILE_OPENED(pTFile)) {
     taosCloseFile(&pTFile->pFile);
-    TD_FILE_SET_CLOSED(pTFile);
+    TD_TFILE_SET_CLOSED(pTFile);
   }
 }
 
@@ -183,8 +187,8 @@ int32_t tdInitTFile(STFile *pTFile, STfs *pTfs, const char *fname) {
   char    fullname[TSDB_FILENAME_LEN];
   SDiskID did = {0};
 
-  TD_FILE_SET_STATE(pTFile, TD_FILE_STATE_OK);
-  TD_FILE_SET_CLOSED(pTFile);
+  TD_TFILE_SET_STATE(pTFile, TD_FILE_STATE_OK);
+  TD_TFILE_SET_CLOSED(pTFile);
 
   memset(&(pTFile->info), 0, sizeof(pTFile->info));
   pTFile->info.magic = TD_FILE_INIT_MAGIC;
@@ -202,18 +206,18 @@ int32_t tdInitTFile(STFile *pTFile, STfs *pTfs, const char *fname) {
 int32_t tdCreateTFile(STFile *pTFile, STfs *pTfs, bool updateHeader, int8_t fType) {
   ASSERT(pTFile->info.fsize == 0 && pTFile->info.magic == TD_FILE_INIT_MAGIC);
 
-  pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pTFile->pFile = taosOpenFile(TD_TFILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
   if (pTFile->pFile == NULL) {
     if (errno == ENOENT) {
       // Try to create directory recursively
-      char *s = strdup(TD_FILE_REL_NAME(pTFile));
-      if (tfsMkdirRecurAt(pTfs, taosDirName(s), TD_FILE_DID(pTFile)) < 0) {
+      char *s = strdup(TD_TFILE_REL_NAME(pTFile));
+      if (tfsMkdirRecurAt(pTfs, taosDirName(s), TD_TFILE_DID(pTFile)) < 0) {
         taosMemoryFreeClear(s);
         return -1;
       }
       taosMemoryFreeClear(s);
 
-      pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+      pTFile->pFile = taosOpenFile(TD_TFILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
       if (pTFile->pFile == NULL) {
         terrno = TAOS_SYSTEM_ERROR(errno);
         return -1;
@@ -240,7 +244,7 @@ int32_t tdCreateTFile(STFile *pTFile, STfs *pTfs, bool updateHeader, int8_t fTyp
   return 0;
 }
 
-int32_t tdRemoveTFile(STFile *pTFile) { return tfsRemoveFile(TD_FILE_F(pTFile)); }
+int32_t tdRemoveTFile(STFile *pTFile) { return tfsRemoveFile(TD_TFILE_F(pTFile)); }
 
 // smaXXXUtil ================
 // ...
\ No newline at end of file

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

@@ -152,14 +152,14 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
   return pVnode;
 
 _err:
-  if (pVnode->pSma) smaClose(pVnode->pSma);
+  if (pVnode->pSma) smaCloseEnv(pVnode->pSma);
   if (pVnode->pQuery) vnodeQueryClose(pVnode);
   if (pVnode->pTq) tqClose(pVnode->pTq);
   if (pVnode->pWal) walClose(pVnode->pWal);
   if (pVnode->pTsdb) tsdbClose(&pVnode->pTsdb);
+  if (pVnode->pSma) smaCloseEx(pVnode->pSma);
   if (pVnode->pMeta) metaClose(pVnode->pMeta);
 
-
   tsem_destroy(&(pVnode->canCommit));
   taosMemoryFree(pVnode);
   return NULL;

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

@@ -243,7 +243,7 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       char          *syncNodeStr = sync2SimpleStr(pVnode->sync);
       static int64_t vndTick = 0;
       if (++vndTick % 10 == 1) {
-        vGTrace("vgId:%d, sync heartbeat msg:%s, %s", syncGetVgId(pVnode->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
+        vGTrace("vgId:%d, sync trace msg:%s, %s", syncGetVgId(pVnode->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
       }
       if (gRaftDetailLog) {
         char logBuf[512] = {0};

source/libs/catalog/src/catalog.c

@@ -1291,7 +1291,9 @@ void catalogDestroy(void) {
 
   atomic_store_8((int8_t*)&gCtgMgmt.exit, true);
 
-  ctgClearCacheEnqueue(NULL, true, true, true);
+  if (!taosCheckCurrentInDll()) {
+    ctgClearCacheEnqueue(NULL, true, true, true);
+  }
 
   taosHashCleanup(gCtgMgmt.pCluster);
   gCtgMgmt.pCluster = NULL;

source/libs/executor/src/executorimpl.c

@@ -273,8 +273,7 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
   }
 
   // 1. close current opened time window
-  if (pResultRowInfo->cur.pageId != -1 && ((pResult == NULL) || (pResult->pageId != pResultRowInfo->cur.pageId &&
-                                                                 pResult->offset != pResultRowInfo->cur.offset))) {
+  if (pResultRowInfo->cur.pageId != -1 && ((pResult == NULL) || (pResult->pageId != pResultRowInfo->cur.pageId))) {
     SResultRowPosition pos = pResultRowInfo->cur;
     SFilePage*         pPage = getBufPage(pResultBuf, pos.pageId);
     releaseBufPage(pResultBuf, pPage);

source/libs/executor/src/scanoperator.c

@@ -1600,13 +1600,11 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
       strncpy(pInfo->req.tb, tNameGetTableName(&pInfo->name), tListLen(pInfo->req.tb));
       strcpy(pInfo->req.user, pInfo->pUser);
 
-/*
       if (pInfo->showRewrite) {
         char dbName[TSDB_DB_NAME_LEN] = {0};
         getDBNameFromCondition(pInfo->pCondition, dbName);
         sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);
       }
-*/
 
       int32_t contLen = tSerializeSRetrieveTableReq(NULL, 0, &pInfo->req);
       char*   buf1 = taosMemoryCalloc(1, contLen);

source/libs/function/inc/builtinsimpl.h

@@ -191,6 +191,11 @@ bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool uniqueFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t uniqueFunction(SqlFunctionCtx *pCtx);
 
+bool getModeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool modeFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t modeFunction(SqlFunctionCtx *pCtx);
+int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
+
 bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool twaFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t twaFunction(SqlFunctionCtx *pCtx);

source/libs/function/src/builtins.c

@@ -1045,20 +1045,29 @@ static int32_t translateFirstLastMerge(SFunctionNode* pFunc, char* pErrBuf, int3
   return translateFirstLastImpl(pFunc, pErrBuf, len, false);
 }
 
-static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+static int32_t translateUniqueMode(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isUnique) {
   if (1 != LIST_LENGTH(pFunc->pParameterList)) {
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
   SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
   if (!nodesExprHasColumn(pPara)) {
-    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, "The parameters of UNIQUE must contain columns");
+    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, "The parameters of %s must contain columns",
+                           isUnique ? "UNIQUE" : "MODE");
   }
 
   pFunc->node.resType = ((SExprNode*)pPara)->resType;
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateUniqueMode(pFunc, pErrBuf, len, true);
+}
+
+static int32_t translateMode(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  return translateUniqueMode(pFunc, pErrBuf, len, false);
+}
+
 static int32_t translateDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList);
   if (numOfParams == 0 || numOfParams > 2) {
@@ -1220,19 +1229,19 @@ static int32_t translateSubstr(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
 
 static int32_t translateCast(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
   // The number of parameters has been limited by the syntax definition
-  //uint8_t para1Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
+  // uint8_t para1Type = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
 
   // The function return type has been set during syntax parsing
   uint8_t para2Type = pFunc->node.resType.type;
-  //if (para2Type != TSDB_DATA_TYPE_BIGINT && para2Type != TSDB_DATA_TYPE_UBIGINT &&
-  //    para2Type != TSDB_DATA_TYPE_VARCHAR && para2Type != TSDB_DATA_TYPE_NCHAR &&
-  //    para2Type != TSDB_DATA_TYPE_TIMESTAMP) {
-  //  return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
-  //}
-  //if ((para2Type == TSDB_DATA_TYPE_TIMESTAMP && IS_VAR_DATA_TYPE(para1Type)) ||
-  //    (para2Type == TSDB_DATA_TYPE_BINARY && para1Type == TSDB_DATA_TYPE_NCHAR)) {
-  //  return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
-  //}
+  // if (para2Type != TSDB_DATA_TYPE_BIGINT && para2Type != TSDB_DATA_TYPE_UBIGINT &&
+  //     para2Type != TSDB_DATA_TYPE_VARCHAR && para2Type != TSDB_DATA_TYPE_NCHAR &&
+  //     para2Type != TSDB_DATA_TYPE_TIMESTAMP) {
+  //   return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  // }
+  // if ((para2Type == TSDB_DATA_TYPE_TIMESTAMP && IS_VAR_DATA_TYPE(para1Type)) ||
+  //     (para2Type == TSDB_DATA_TYPE_BINARY && para1Type == TSDB_DATA_TYPE_NCHAR)) {
+  //   return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  // }
 
   int32_t para2Bytes = pFunc->node.resType.bytes;
   if (IS_VAR_DATA_TYPE(para2Type)) {
@@ -1882,7 +1891,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "first",
     .type = FUNCTION_TYPE_FIRST,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateFirstLast,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -1917,7 +1926,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "last",
     .type = FUNCTION_TYPE_LAST,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
+    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
     .translateFunc = translateFirstLast,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -2109,7 +2118,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "unique",
     .type = FUNCTION_TYPE_UNIQUE,
-    .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | 
+    .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | 
                       FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC,
     .translateFunc = translateUnique,
     .getEnvFunc   = getUniqueFuncEnv,
@@ -2117,6 +2126,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = uniqueFunction,
     .finalizeFunc = NULL
   },
+  {
+    .name = "mode",
+    .type = FUNCTION_TYPE_MODE,
+    .classification = FUNC_MGT_AGG_FUNC,
+    .translateFunc = translateMode,
+    .getEnvFunc   = getModeFuncEnv,
+    .initFunc     = modeFunctionSetup,
+    .processFunc  = modeFunction,
+    .finalizeFunc = modeFinalize,
+  },
   {
     .name = "abs",
     .type = FUNCTION_TYPE_ABS,

source/libs/function/src/builtinsimpl.c

@@ -32,6 +32,7 @@
 #define TAIL_MAX_OFFSET        100
 
 #define UNIQUE_MAX_RESULT_SIZE (1024 * 1024 * 10)
+#define MODE_MAX_RESULT_SIZE   UNIQUE_MAX_RESULT_SIZE
 
 #define HLL_BUCKET_BITS 14  // The bits of the bucket
 #define HLL_DATA_BITS   (64 - HLL_BUCKET_BITS)
@@ -246,6 +247,19 @@ typedef struct SUniqueInfo {
   char      pItems[];
 } SUniqueInfo;
 
+typedef struct SModeItem {
+  int64_t count;
+  char    data[];
+} SModeItem;
+
+typedef struct SModeInfo {
+  int32_t   numOfPoints;
+  uint8_t   colType;
+  int16_t   colBytes;
+  SHashObj* pHash;
+  char      pItems[];
+} SModeInfo;
+
 typedef struct SDerivInfo {
   double  prevValue;       // previous value
   TSKEY   prevTs;          // previous timestamp
@@ -4694,21 +4708,100 @@ int32_t uniqueFunction(SqlFunctionCtx* pCtx) {
   return pInfo->numOfPoints;
 }
 
-int32_t uniqueFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+bool getModeFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(SModeInfo) + MODE_MAX_RESULT_SIZE;
+  return true;
+}
+
+bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
+  if (!functionSetup(pCtx, pResInfo)) {
+    return false;
+  }
+
+  SModeInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+  pInfo->numOfPoints = 0;
+  pInfo->colType = pCtx->resDataInfo.type;
+  pInfo->colBytes = pCtx->resDataInfo.bytes;
+  if (pInfo->pHash != NULL) {
+    taosHashClear(pInfo->pHash);
+  } else {
+    pInfo->pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  }
+  return true;
+}
+
+static void doModeAdd(SModeInfo* pInfo, char* data, bool isNull) {
+  // ignore null elements
+  if (isNull) {
+    return;
+  }
+
+  int32_t      hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
+  SModeItem**     pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
+  if (pHashItem == NULL) {
+    int32_t      size = sizeof(SModeItem) + pInfo->colBytes;
+    SModeItem* pItem  = (SModeItem*)(pInfo->pItems + pInfo->numOfPoints * size);
+    memcpy(pItem->data, data, pInfo->colBytes);
+    pItem->count += 1;
+
+    taosHashPut(pInfo->pHash, data, hashKeyBytes, &pItem, sizeof(SModeItem*));
+    pInfo->numOfPoints++;
+  } else {
+    (*pHashItem)->count += 1;
+  }
+}
+
+int32_t modeFunction(SqlFunctionCtx* pCtx) {
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
-  SUniqueInfo*         pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+  SModeInfo*              pInfo = GET_ROWCELL_INTERBUF(pResInfo);
+
+  SInputColumnInfoData* pInput = &pCtx->input;
+
+  SColumnInfoData* pInputCol = pInput->pData[0];
+  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
+
+  int32_t startOffset = pCtx->offset;
+  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
+    char* data = colDataGetData(pInputCol, i);
+    doModeAdd(pInfo, data, colDataIsNull_s(pInputCol, i));
+
+    if (sizeof(SModeInfo) + pInfo->numOfPoints * (sizeof(SModeItem) + pInfo->colBytes) >= MODE_MAX_RESULT_SIZE) {
+      taosHashCleanup(pInfo->pHash);
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
+  SET_VAL(pResInfo, 1, 1);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
+  SModeInfo*           pInfo = GET_ROWCELL_INTERBUF(pResInfo);
   int32_t              slotId = pCtx->pExpr->base.resSchema.slotId;
   SColumnInfoData*     pCol = taosArrayGet(pBlock->pDataBlock, slotId);
+  int32_t              currentRow = pBlock->info.rows;
 
-  for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
-    SUniqueItem* pItem = (SUniqueItem*)(pInfo->pItems + i * (sizeof(SUniqueItem) + pInfo->colBytes));
-    colDataAppend(pCol, i, pItem->data, false);
-    // TODO: handle ts output
+  int32_t resIndex;
+  int32_t maxCount = 0;
+  for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
+    SModeItem* pItem = (SModeItem*)(pInfo->pItems + i * (sizeof(SModeItem) + pInfo->colBytes));
+    if (pItem->count > maxCount) {
+      maxCount = pItem->count;
+      resIndex = i;
+    } else if (pItem->count == maxCount) {
+      resIndex = -1;
+    }
   }
 
+  SModeItem* pResItem = (SModeItem*)(pInfo->pItems + resIndex * (sizeof(SModeItem) + pInfo->colBytes));
+  colDataAppend(pCol, currentRow, pResItem->data, (resIndex == -1) ? true : false);
+
   return pResInfo->numOfRes;
 }
 
+
 bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(STwaInfo);
   return true;

source/libs/function/test/udf1.c

@@ -2,12 +2,8 @@
 #include <stdlib.h>
 #include <stdio.h>
 
-#include "tudf.h"
+#include "taosudf.h"
 
-#undef malloc
-#define malloc malloc
-#undef free
-#define free free
 
 DLL_EXPORT int32_t udf1_init() {
   return 0;

source/libs/function/test/udf2.c

 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
+#include <math.h>
 
-#include "tudf.h"
-
-#undef malloc
-#define malloc malloc
-#undef free
-#define free free
+#include "taosudf.h"
 
 DLL_EXPORT int32_t udf2_init() {
   return 0;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -100,6 +100,7 @@ static int32_t exprNodeCopy(const SExprNode* pSrc, SExprNode* pDst) {
   COPY_OBJECT_FIELD(resType, sizeof(SDataType));
   COPY_CHAR_ARRAY_FIELD(aliasName);
   COPY_CHAR_ARRAY_FIELD(userAlias);
+  COPY_SCALAR_FIELD(orderAlias);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/nodes/src/nodesUtilFuncs.c

@@ -1500,7 +1500,8 @@ typedef struct SCollectFuncsCxt {
 
 static EDealRes collectFuncs(SNode* pNode, void* pContext) {
   SCollectFuncsCxt* pCxt = (SCollectFuncsCxt*)pContext;
-  if (QUERY_NODE_FUNCTION == nodeType(pNode) && pCxt->classifier(((SFunctionNode*)pNode)->funcId)) {
+  if (QUERY_NODE_FUNCTION == nodeType(pNode) && pCxt->classifier(((SFunctionNode*)pNode)->funcId) &&
+      !(((SExprNode*)pNode)->orderAlias)) {
     pCxt->errCode = nodesListStrictAppend(pCxt->pFuncs, nodesCloneNode(pNode));
     return (TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR);
   }

source/libs/parser/src/parTranslater.c

@@ -1355,25 +1355,6 @@ static EDealRes rewriteColToSelectValFunc(STranslateContext* pCxt, SNode** pNode
   return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR;
 }
 
-static EDealRes rewriteExprToGroupKeyFunc(STranslateContext* pCxt, SNode** pNode) {
-  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
-  if (NULL == pFunc) {
-    pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
-    return DEAL_RES_ERROR;
-  }
-
-  strcpy(pFunc->functionName, "_group_key");
-  strcpy(pFunc->node.aliasName, ((SExprNode*)*pNode)->aliasName);
-  pCxt->errCode = nodesListMakeAppend(&pFunc->pParameterList, *pNode);
-  if (TSDB_CODE_SUCCESS == pCxt->errCode) {
-    *pNode = (SNode*)pFunc;
-    pCxt->errCode = fmGetFuncInfo(pFunc, pCxt->msgBuf.buf, pCxt->msgBuf.len);
-  }
-  pCxt->pCurrSelectStmt->hasAggFuncs = true;
-
-  return (TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR);
-}
-
 static EDealRes doCheckExprForGroupBy(SNode** pNode, void* pContext) {
   SCheckExprForGroupByCxt* pCxt = (SCheckExprForGroupByCxt*)pContext;
   if (!nodesIsExprNode(*pNode) || isAliasColumn(*pNode)) {
@@ -1393,7 +1374,13 @@ static EDealRes doCheckExprForGroupBy(SNode** pNode, void* pContext) {
   SNode* pGroupNode = NULL;
   FOREACH(pGroupNode, getGroupByList(pCxt->pTranslateCxt)) {
     if (nodesEqualNode(getGroupByNode(pGroupNode), *pNode)) {
-      return rewriteExprToGroupKeyFunc(pCxt->pTranslateCxt, pNode);
+      return DEAL_RES_IGNORE_CHILD;
+    }
+  }
+  SNode* pPartKey = NULL;
+  FOREACH(pPartKey, pCxt->pTranslateCxt->pCurrSelectStmt->pPartitionByList) {
+    if (nodesEqualNode(pPartKey, *pNode)) {
+      return DEAL_RES_IGNORE_CHILD;
     }
   }
   if (isScanPseudoColumnFunc(*pNode) || QUERY_NODE_COLUMN == nodeType(*pNode)) {
@@ -1451,25 +1438,6 @@ static int32_t rewriteColsToSelectValFunc(STranslateContext* pCxt, SSelectStmt*
   return pCxt->errCode;
 }
 
-static EDealRes rewriteExprsToGroupKeyFuncImpl(SNode** pNode, void* pContext) {
-  STranslateContext* pCxt = pContext;
-  SNode*             pPartKey = NULL;
-  FOREACH(pPartKey, pCxt->pCurrSelectStmt->pPartitionByList) {
-    if (nodesEqualNode(pPartKey, *pNode)) {
-      return rewriteExprToGroupKeyFunc(pCxt, pNode);
-    }
-  }
-  return DEAL_RES_CONTINUE;
-}
-
-static int32_t rewriteExprsToGroupKeyFunc(STranslateContext* pCxt, SSelectStmt* pSelect) {
-  nodesRewriteExprs(pSelect->pProjectionList, rewriteExprsToGroupKeyFuncImpl, pCxt);
-  if (TSDB_CODE_SUCCESS == pCxt->errCode && !pSelect->isDistinct) {
-    nodesRewriteExprs(pSelect->pOrderByList, rewriteExprsToGroupKeyFuncImpl, pCxt);
-  }
-  return pCxt->errCode;
-}
-
 typedef struct CheckAggColCoexistCxt {
   STranslateContext* pTranslateCxt;
   bool               existAggFunc;
@@ -1529,9 +1497,6 @@ static int32_t checkAggColCoexist(STranslateContext* pCxt, SSelectStmt* pSelect)
   if (cxt.existIndefiniteRowsFunc && cxt.existCol) {
     return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
   }
-  if (cxt.existAggFunc && NULL != pSelect->pPartitionByList) {
-    return rewriteExprsToGroupKeyFunc(pCxt, pSelect);
-  }
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2408,54 +2373,6 @@ static EDealRes rewriteSeletcValueFunc(STranslateContext* pCxt, SNode** pNode) {
   return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR;
 }
 
-static EDealRes rewriteUniqueFunc(SNode** pNode, void* pContext) {
-  SRwriteUniqueCxt* pCxt = pContext;
-  if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
-    SFunctionNode* pFunc = (SFunctionNode*)*pNode;
-    if (FUNCTION_TYPE_UNIQUE == pFunc->funcType) {
-      SNode* pExpr = nodesListGetNode(pFunc->pParameterList, 0);
-      NODES_CLEAR_LIST(pFunc->pParameterList);
-      strcpy(((SExprNode*)pExpr)->aliasName, ((SExprNode*)*pNode)->aliasName);
-      nodesDestroyNode(*pNode);
-      *pNode = pExpr;
-      pCxt->pExpr = pExpr;
-      return DEAL_RES_IGNORE_CHILD;
-    } else if (FUNCTION_TYPE_SELECT_VALUE == pFunc->funcType) {
-      return rewriteSeletcValueFunc(pCxt->pTranslateCxt, pNode);
-    }
-  }
-  return DEAL_RES_CONTINUE;
-}
-
-static SNode* createGroupingSet(SNode* pExpr) {
-  SGroupingSetNode* pGroupingSet = (SGroupingSetNode*)nodesMakeNode(QUERY_NODE_GROUPING_SET);
-  if (NULL == pGroupingSet) {
-    return NULL;
-  }
-  pGroupingSet->groupingSetType = GP_TYPE_NORMAL;
-  if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pGroupingSet->pParameterList, nodesCloneNode(pExpr))) {
-    nodesDestroyNode((SNode*)pGroupingSet);
-    return NULL;
-  }
-  return (SNode*)pGroupingSet;
-}
-
-// from: select unique(expr), col1 + col2 from t where_clause partition_by_clause order_by_clause ...
-// to: select expr, first(col1) + first(col2) from t where_clause partition_by_clause group by expr order_by_clause ...
-static int32_t rewriteUniqueStmt(STranslateContext* pCxt, SSelectStmt* pSelect) {
-  if (!pSelect->hasUniqueFunc) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  SRwriteUniqueCxt cxt = {.pTranslateCxt = pCxt, .pExpr = NULL};
-  nodesRewriteExprs(pSelect->pProjectionList, rewriteUniqueFunc, &cxt);
-  if (TSDB_CODE_SUCCESS == cxt.pTranslateCxt->errCode) {
-    cxt.pTranslateCxt->errCode = nodesListMakeStrictAppend(&pSelect->pGroupByList, createGroupingSet(cxt.pExpr));
-  }
-  pSelect->hasIndefiniteRowsFunc = false;
-  return cxt.pTranslateCxt->errCode;
-}
-
 typedef struct SReplaceOrderByAliasCxt {
   STranslateContext* pTranslateCxt;
   SNodeList*         pProjectionList;
@@ -2474,6 +2391,7 @@ static EDealRes replaceOrderByAliasImpl(SNode** pNode, void* pContext) {
           pCxt->pTranslateCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
           return DEAL_RES_ERROR;
         }
+        ((SExprNode*)pNew)->orderAlias = true;
         nodesDestroyNode(*pNode);
         *pNode = pNew;
         return DEAL_RES_CONTINUE;
@@ -2529,9 +2447,6 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
   if (TSDB_CODE_SUCCESS == code) {
     code = translateInterp(pCxt, pSelect);
   }
-  if (TSDB_CODE_SUCCESS == code) {
-    code = rewriteUniqueStmt(pCxt, pSelect);
-  }
   if (TSDB_CODE_SUCCESS == code) {
     code = rewriteTimelineFunc(pCxt, pSelect);
   }

source/libs/parser/test/parSelectTest.cpp

@@ -232,8 +232,6 @@ TEST_F(ParserSelectTest, groupBySemanticCheck) {
 
   run("SELECT COUNT(*) cnt, c1 FROM t1 WHERE c1 > 0", TSDB_CODE_PAR_NOT_SINGLE_GROUP);
   run("SELECT COUNT(*) cnt, c2 FROM t1 WHERE c1 > 0 GROUP BY c1", TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION);
-  run("SELECT COUNT(*) cnt, c2 FROM t1 WHERE c1 > 0 PARTITION BY c2 GROUP BY c1",
-      TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION);
 }
 
 TEST_F(ParserSelectTest, orderBy) {

source/libs/planner/src/planLogicCreater.c

@@ -437,6 +437,33 @@ static SColumnNode* createColumnByExpr(const char* pStmtName, SExprNode* pExpr)
   return pCol;
 }
 
+static SNode* createGroupingSetNode(SNode* pExpr) {
+  SGroupingSetNode* pGroupingSet = (SGroupingSetNode*)nodesMakeNode(QUERY_NODE_GROUPING_SET);
+  if (NULL == pGroupingSet) {
+    return NULL;
+  }
+  pGroupingSet->groupingSetType = GP_TYPE_NORMAL;
+  if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pGroupingSet->pParameterList, nodesCloneNode(pExpr))) {
+    nodesDestroyNode((SNode*)pGroupingSet);
+    return NULL;
+  }
+  return (SNode*)pGroupingSet;
+}
+
+static int32_t createGroupKeysFromPartKeys(SNodeList* pPartKeys, SNodeList** pOutput) {
+  SNodeList* pGroupKeys = NULL;
+  SNode*     pPartKey = NULL;
+  FOREACH(pPartKey, pPartKeys) {
+    int32_t code = nodesListMakeStrictAppend(&pGroupKeys, createGroupingSetNode(pPartKey));
+    if (TSDB_CODE_SUCCESS != code) {
+      nodesDestroyList(pGroupKeys);
+      return code;
+    }
+  }
+  *pOutput = pGroupKeys;
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
   if (!pSelect->hasAggFuncs && NULL == pSelect->pGroupByList) {
     return TSDB_CODE_SUCCESS;
@@ -459,10 +486,18 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
     code = rewriteExprsForSelect(pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY);
   }
 
+  if (NULL != pSelect->pPartitionByList) {
+    code = createGroupKeysFromPartKeys(pSelect->pPartitionByList, &pAgg->pGroupKeys);
+  }
+
   if (NULL != pSelect->pGroupByList) {
-    pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
-    if (NULL == pAgg->pGroupKeys) {
-      code = TSDB_CODE_OUT_OF_MEMORY;
+    if (NULL != pAgg->pGroupKeys) {
+      code = nodesListStrictAppendList(pAgg->pGroupKeys, nodesCloneList(pSelect->pGroupByList));
+    } else {
+      pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
+      if (NULL == pAgg->pGroupKeys) {
+        code = TSDB_CODE_OUT_OF_MEMORY;
+      }
     }
   }
 
@@ -508,6 +543,7 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
   }
 
   pIdfRowsFunc->isTailFunc = pSelect->hasTailFunc;
+  pIdfRowsFunc->isUniqueFunc = pSelect->hasUniqueFunc;
 
   // indefinite rows functions and _select_values functions
   int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
@@ -809,7 +845,8 @@ static int32_t createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSel
 }
 
 static int32_t createPartitionLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
-  if (NULL == pSelect->pPartitionByList) {
+  if (NULL == pSelect->pPartitionByList || (pSelect->hasAggFuncs && NULL == pSelect->pWindow) ||
+      NULL != pSelect->pGroupByList) {
     return TSDB_CODE_SUCCESS;
   }
 

source/libs/planner/src/planOptimizer.c

@@ -987,6 +987,7 @@ static int32_t smaIndexOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogi
       code = smaIndexOptApplyIndex(pLogicSubplan, pScan, pIndex, pSmaCols, wstrartIndex);
       taosArrayDestroyEx(pScan->pSmaIndexes, smaIndexOptDestroySmaIndex);
       pScan->pSmaIndexes = NULL;
+      pCxt->optimized = true;
       break;
     }
   }
@@ -1033,12 +1034,30 @@ static SNodeList* partTagsGetPartKeys(SLogicNode* pNode) {
   }
 }
 
+static SNodeList* partTagsGetFuncs(SLogicNode* pNode) {
+  if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
+    return NULL;
+  } else {
+    return ((SAggLogicNode*)pNode)->pAggFuncs;
+  }
+}
+
+static bool partTagsOptAreSupportedFuncs(SNodeList* pFuncs) {
+  SNode* pFunc = NULL;
+  FOREACH(pFunc, pFuncs) {
+    if (fmIsIndefiniteRowsFunc(((SFunctionNode*)pFunc)->funcId) && !fmIsSelectFunc(((SFunctionNode*)pFunc)->funcId)) {
+      return false;
+    }
+  }
+  return true;
+}
+
 static bool partTagsOptMayBeOptimized(SLogicNode* pNode) {
   if (!partTagsIsOptimizableNode(pNode)) {
     return false;
   }
 
-  return !partTagsOptHasCol(partTagsGetPartKeys(pNode));
+  return !partTagsOptHasCol(partTagsGetPartKeys(pNode)) && partTagsOptAreSupportedFuncs(partTagsGetFuncs(pNode));
 }
 
 static EDealRes partTagsOptRebuildTbanmeImpl(SNode** pNode, void* pContext) {
@@ -1064,6 +1083,59 @@ static int32_t partTagsOptRebuildTbanme(SNodeList* pPartKeys) {
   return code;
 }
 
+static SNode* partTagsCreateWrapperFunc(const char* pFuncName, SNode* pNode) {
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFunc) {
+    return NULL;
+  }
+
+  strcpy(pFunc->functionName, pFuncName);
+  if (QUERY_NODE_COLUMN == nodeType(pNode)) {
+    SColumnNode* pCol = (SColumnNode*)pNode;
+    sprintf(pFunc->node.aliasName, "%s.%s", pCol->tableAlias, pCol->colName);
+  } else {
+    strcpy(pFunc->node.aliasName, ((SExprNode*)pNode)->aliasName);
+  }
+  int32_t code = nodesListMakeStrictAppend(&pFunc->pParameterList, nodesCloneNode(pNode));
+  if (TSDB_CODE_SUCCESS == code) {
+    code = fmGetFuncInfo(pFunc, NULL, 0);
+  }
+
+  if (TSDB_CODE_SUCCESS != code) {
+    nodesDestroyNode((SNode*)pFunc);
+    return NULL;
+  }
+
+  return (SNode*)pFunc;
+}
+
+static bool partTagsHasIndefRowsSelectFunc(SNodeList* pFuncs) {
+  SNode* pFunc = NULL;
+  FOREACH(pFunc, pFuncs) {
+    if (fmIsIndefiniteRowsFunc(((SFunctionNode*)pFunc)->funcId)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, SNodeList* pAggFuncs) {
+  bool    hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAggFuncs);
+  int32_t code = TSDB_CODE_SUCCESS;
+  SNode*  pNode = NULL;
+  FOREACH(pNode, pGroupTags) {
+    if (hasIndefRowsSelectFunc) {
+      code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
+    } else {
+      code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_group_key", pNode));
+    }
+    if (TSDB_CODE_SUCCESS != code) {
+      break;
+    }
+  }
+  return code;
+}
+
 static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
   SLogicNode* pNode = optFindPossibleNode(pLogicSubplan->pNode, partTagsOptMayBeOptimized);
   if (NULL == pNode) {
@@ -1080,15 +1152,17 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
       nodesDestroyNode((SNode*)pNode);
     }
   } else {
-    SNode* pGroupKey = NULL;
-    FOREACH(pGroupKey, ((SAggLogicNode*)pNode)->pGroupKeys) {
+    SAggLogicNode* pAgg = (SAggLogicNode*)pNode;
+    SNode*         pGroupKey = NULL;
+    FOREACH(pGroupKey, pAgg->pGroupKeys) {
       code = nodesListMakeStrictAppend(
           &pScan->pGroupTags, nodesCloneNode(nodesListGetNode(((SGroupingSetNode*)pGroupKey)->pParameterList, 0)));
       if (TSDB_CODE_SUCCESS != code) {
         break;
       }
     }
-    NODES_DESTORY_LIST(((SAggLogicNode*)pNode)->pGroupKeys);
+    NODES_DESTORY_LIST(pAgg->pGroupKeys);
+    code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, pAgg->pAggFuncs);
   }
   if (TSDB_CODE_SUCCESS == code) {
     code = partTagsOptRebuildTbanme(pScan->pGroupTags);
@@ -1125,7 +1199,8 @@ static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
   }
 
   SProjectLogicNode* pProjectNode = (SProjectLogicNode*)pNode;
-  if (NULL != pProjectNode->node.pLimit || NULL != pProjectNode->node.pSlimit) {
+  if (NULL != pProjectNode->node.pLimit || NULL != pProjectNode->node.pSlimit ||
+      NULL != pProjectNode->node.pConditions) {
     return false;
   }
 
@@ -1163,6 +1238,7 @@ static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
     NODES_CLEAR_LIST(pProjectNode->node.pChildren);
     nodesDestroyNode((SNode*)pProjectNode);
   }
+  pCxt->optimized = true;
   return code;
 }
 
@@ -1306,6 +1382,7 @@ static int32_t rewriteTailOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pL
     nodesDestroyNode((SNode*)pSort);
     nodesDestroyNode((SNode*)pProject);
   }
+  pCxt->optimized = true;
   return code;
 }
 
@@ -1344,6 +1421,7 @@ static int32_t eliminateSetOpOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
 
       pSetOpNode->pChildren = NULL;
       ERASE_NODE(pSetOpNode->pParent->pChildren);
+      pCxt->optimized = true;
       return TSDB_CODE_SUCCESS;
     }
   }
@@ -1360,6 +1438,261 @@ static int32_t eliminateSetOpOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLo
   return eliminateSetOpOptimizeImpl(pCxt, pLogicSubplan, pSetOpNode);
 }
 
+static bool rewriteUniqueOptMayBeOptimized(SLogicNode* pNode) {
+  return QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC == nodeType(pNode) && ((SIndefRowsFuncLogicNode*)pNode)->isUniqueFunc;
+}
+
+static SNode* rewriteUniqueOptCreateGroupingSet(SNode* pExpr) {
+  SGroupingSetNode* pGroupingSet = (SGroupingSetNode*)nodesMakeNode(QUERY_NODE_GROUPING_SET);
+  if (NULL == pGroupingSet) {
+    return NULL;
+  }
+  pGroupingSet->groupingSetType = GP_TYPE_NORMAL;
+  SExprNode* pGroupExpr = (SExprNode*)nodesCloneNode(pExpr);
+  if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pGroupingSet->pParameterList, (SNode*)pGroupExpr)) {
+    nodesDestroyNode((SNode*)pGroupingSet);
+    return NULL;
+  }
+  return (SNode*)pGroupingSet;
+}
+
+static SNode* rewriteUniqueOptCreateFirstFunc(SFunctionNode* pSelectValue, SNode* pCol) {
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
+  if (NULL == pFunc) {
+    return NULL;
+  }
+
+  strcpy(pFunc->functionName, "first");
+  if (NULL != pSelectValue) {
+    sprintf(pFunc->node.aliasName, "%s", pSelectValue->node.aliasName);
+  } else {
+    sprintf(pFunc->node.aliasName, "%s.%p", pFunc->functionName, pFunc);
+  }
+  int32_t code = nodesListMakeStrictAppend(&pFunc->pParameterList, nodesCloneNode(pCol));
+  if (TSDB_CODE_SUCCESS == code) {
+    code = fmGetFuncInfo(pFunc, NULL, 0);
+  }
+
+  if (TSDB_CODE_SUCCESS != code) {
+    nodesDestroyNode((SNode*)pFunc);
+    return NULL;
+  }
+
+  return (SNode*)pFunc;
+}
+
+static int32_t rewriteUniqueOptCreateAgg(SIndefRowsFuncLogicNode* pIndef, SLogicNode** pOutput) {
+  SAggLogicNode* pAgg = (SAggLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_AGG);
+  if (NULL == pAgg) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  TSWAP(pAgg->node.pChildren, pIndef->node.pChildren);
+  pAgg->node.precision = pIndef->node.precision;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  bool    hasSelectPrimaryKey = false;
+  SNode*  pPrimaryKey = NULL;
+  SNode*  pNode = NULL;
+  FOREACH(pNode, pIndef->pFuncs) {
+    SFunctionNode* pFunc = (SFunctionNode*)pNode;
+    SNode*         pExpr = nodesListGetNode(pFunc->pParameterList, 0);
+    if (FUNCTION_TYPE_UNIQUE == pFunc->funcType) {
+      pPrimaryKey = nodesListGetNode(pFunc->pParameterList, 1);
+      code = nodesListMakeStrictAppend(&pAgg->pGroupKeys, rewriteUniqueOptCreateGroupingSet(pExpr));
+    } else if (PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pExpr)->colId) {  // _select_value(ts) => first(ts)
+      hasSelectPrimaryKey = true;
+      code = nodesListMakeStrictAppend(&pAgg->pAggFuncs, rewriteUniqueOptCreateFirstFunc(pFunc, pExpr));
+    } else {  // _select_value(other_col)
+      code = nodesListMakeStrictAppend(&pAgg->pAggFuncs, nodesCloneNode(pNode));
+    }
+    if (TSDB_CODE_SUCCESS != code) {
+      break;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    code = createColumnByRewriteExprs(pAgg->pGroupKeys, &pAgg->node.pTargets);
+  }
+  if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
+    code = createColumnByRewriteExprs(pAgg->pAggFuncs, &pAgg->node.pTargets);
+  }
+
+  if (TSDB_CODE_SUCCESS == code && !hasSelectPrimaryKey && NULL != pAgg->pAggFuncs) {
+    code = nodesListMakeStrictAppend(&pAgg->pAggFuncs, rewriteUniqueOptCreateFirstFunc(NULL, pPrimaryKey));
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pOutput = (SLogicNode*)pAgg;
+  } else {
+    nodesDestroyNode((SNode*)pAgg);
+  }
+  return code;
+}
+
+static SNode* rewriteUniqueOptCreateProjectCol(SFunctionNode* pFunc) {
+  SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
+  if (NULL == pCol) {
+    return NULL;
+  }
+
+  pCol->node.resType = pFunc->node.resType;
+  if (FUNCTION_TYPE_UNIQUE == pFunc->funcType) {
+    SExprNode* pExpr = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
+    if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
+      strcpy(pCol->tableAlias, ((SColumnNode*)pExpr)->tableAlias);
+      strcpy(pCol->colName, ((SColumnNode*)pExpr)->colName);
+    } else {
+      strcpy(pCol->colName, pExpr->aliasName);
+    }
+  } else {
+    strcpy(pCol->colName, pFunc->node.aliasName);
+  }
+  strcpy(pCol->node.aliasName, pFunc->node.aliasName);
+
+  return (SNode*)pCol;
+}
+
+static int32_t rewriteUniqueOptCreateProject(SIndefRowsFuncLogicNode* pIndef, SLogicNode** pOutput) {
+  SProjectLogicNode* pProject = (SProjectLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_PROJECT);
+  if (NULL == pProject) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  TSWAP(pProject->node.pTargets, pIndef->node.pTargets);
+  pProject->node.precision = pIndef->node.precision;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  SNode*  pNode = NULL;
+  FOREACH(pNode, pIndef->pFuncs) {
+    code = nodesListMakeStrictAppend(&pProject->pProjections, rewriteUniqueOptCreateProjectCol((SFunctionNode*)pNode));
+    if (TSDB_CODE_SUCCESS != code) {
+      break;
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pOutput = (SLogicNode*)pProject;
+  } else {
+    nodesDestroyNode((SNode*)pProject);
+  }
+  return code;
+}
+
+static int32_t rewriteUniqueOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan,
+                                         SIndefRowsFuncLogicNode* pIndef) {
+  SLogicNode* pAgg = NULL;
+  SLogicNode* pProject = NULL;
+  int32_t     code = rewriteUniqueOptCreateAgg(pIndef, &pAgg);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = rewriteUniqueOptCreateProject(pIndef, &pProject);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesListMakeAppend(&pProject->pChildren, (SNode*)pAgg);
+    pAgg->pParent = pProject;
+    pAgg = NULL;
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pIndef, pProject);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    nodesDestroyNode((SNode*)pIndef);
+  } else {
+    nodesDestroyNode((SNode*)pAgg);
+    nodesDestroyNode((SNode*)pProject);
+  }
+  pCxt->optimized = true;
+  return code;
+}
+
+static int32_t rewriteUniqueOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
+  SIndefRowsFuncLogicNode* pIndef =
+      (SIndefRowsFuncLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, rewriteUniqueOptMayBeOptimized);
+
+  if (NULL == pIndef) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  return rewriteUniqueOptimizeImpl(pCxt, pLogicSubplan, pIndef);
+}
+
+// merge projects
+static bool mergeProjectsMayBeOptimized(SLogicNode* pNode) {
+  if (QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(pNode) || 1 != LIST_LENGTH(pNode->pChildren)) {
+    return false;
+  }
+  SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pNode->pChildren, 0);
+  if (QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(pChild) || 1 < LIST_LENGTH(pChild->pChildren) ||
+      NULL != pChild->pConditions || NULL != pChild->pLimit || NULL != pChild->pSlimit) {
+    return false;
+  }
+  return true;
+}
+
+typedef struct SMergeProjectionsContext {
+  SProjectLogicNode* pChildProj;
+  int32_t            errCode;
+} SMergeProjectionsContext;
+
+static EDealRes mergeProjectionsExpr(SNode** pNode, void* pContext) {
+  SMergeProjectionsContext* pCxt = pContext;
+  SProjectLogicNode*        pChildProj = pCxt->pChildProj;
+  if (QUERY_NODE_COLUMN == nodeType(*pNode)) {
+    SNode* pTarget;
+    FOREACH(pTarget, ((SLogicNode*)pChildProj)->pTargets) {
+      if (nodesEqualNode(pTarget, *pNode)) {
+        SNode* pProjection;
+        FOREACH(pProjection, pChildProj->pProjections) {
+          if (0 == strcmp(((SColumnNode*)pTarget)->colName, ((SExprNode*)pProjection)->aliasName)) {
+            SNode* pExpr = nodesCloneNode(pProjection);
+            if (pExpr == NULL) {
+              pCxt->errCode = terrno;
+              return DEAL_RES_ERROR;
+            }
+            nodesDestroyNode(*pNode);
+            *pNode = pExpr;
+          }
+        }
+      }
+    }
+    return DEAL_RES_IGNORE_CHILD;
+  }
+  return DEAL_RES_CONTINUE;
+}
+
+static int32_t mergeProjectsOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan, SLogicNode* pSelfNode) {
+  SLogicNode*              pChild = (SLogicNode*)nodesListGetNode(pSelfNode->pChildren, 0);
+  SMergeProjectionsContext cxt = {.pChildProj = (SProjectLogicNode*)pChild, .errCode = TSDB_CODE_SUCCESS};
+
+  nodesRewriteExprs(((SProjectLogicNode*)pSelfNode)->pProjections, mergeProjectionsExpr, &cxt);
+  int32_t code = cxt.errCode;
+
+  if (TSDB_CODE_SUCCESS == code) {
+    if (1 == LIST_LENGTH(pChild->pChildren)) {
+      SLogicNode* pGrandChild = (SLogicNode*)nodesListGetNode(pChild->pChildren, 0);
+      code = replaceLogicNode(pLogicSubplan, pChild, pGrandChild);
+    } else {  // no grand child
+      NODES_CLEAR_LIST(pSelfNode->pChildren);
+    }
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
+    NODES_CLEAR_LIST(pChild->pChildren);
+  }
+  nodesDestroyNode((SNode*)pChild);
+  pCxt->optimized = true;
+  return code;
+}
+
+static int32_t mergeProjectsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
+  SLogicNode* pProjectNode = optFindPossibleNode(pLogicSubplan->pNode, mergeProjectsMayBeOptimized);
+  if (NULL == pProjectNode) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  return mergeProjectsOptimizeImpl(pCxt, pLogicSubplan, pProjectNode);
+}
+
 // clang-format off
 static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "ScanPath",                   .optimizeFunc = scanPathOptimize},
@@ -1367,25 +1700,40 @@ static const SOptimizeRule optimizeRuleSet[] = {
   {.pName = "SortPrimaryKey",             .optimizeFunc = sortPrimaryKeyOptimize},
   {.pName = "SmaIndex",                   .optimizeFunc = smaIndexOptimize},
   {.pName = "PartitionTags",              .optimizeFunc = partTagsOptimize},
+  {.pName = "MergeProjects",              .optimizeFunc = mergeProjectsOptimize},
   {.pName = "EliminateProject",           .optimizeFunc = eliminateProjOptimize},
   {.pName = "EliminateSetOperator",       .optimizeFunc = eliminateSetOpOptimize},
-  {.pName = "RewriteTail",                .optimizeFunc = rewriteTailOptimize}
+  {.pName = "RewriteTail",                .optimizeFunc = rewriteTailOptimize},
+  {.pName = "RewriteUnique",              .optimizeFunc = rewriteUniqueOptimize}
 };
 // clang-format on
 
 static const int32_t optimizeRuleNum = (sizeof(optimizeRuleSet) / sizeof(SOptimizeRule));
 
+static void dumpLogicSubplan(const char* pRuleName, SLogicSubplan* pSubplan) {
+  char* pStr = NULL;
+  nodesNodeToString((SNode*)pSubplan, false, &pStr, NULL);
+  qDebugL("apply optimize %s rule: %s", pRuleName, pStr);
+  taosMemoryFree(pStr);
+}
+
 static int32_t applyOptimizeRule(SPlanContext* pCxt, SLogicSubplan* pLogicSubplan) {
   SOptimizeContext cxt = {.pPlanCxt = pCxt, .optimized = false};
+  bool             optimized = false;
   do {
-    cxt.optimized = false;
+    optimized = false;
     for (int32_t i = 0; i < optimizeRuleNum; ++i) {
+      cxt.optimized = false;
       int32_t code = optimizeRuleSet[i].optimizeFunc(&cxt, pLogicSubplan);
       if (TSDB_CODE_SUCCESS != code) {
         return code;
       }
+      if (cxt.optimized) {
+        optimized = true;
+        dumpLogicSubplan(optimizeRuleSet[i].pName, pLogicSubplan);
+      }
     }
-  } while (cxt.optimized);
+  } while (optimized);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/src/planSpliter.c

@@ -1237,7 +1237,7 @@ static const int32_t splitRuleNum = (sizeof(splitRuleSet) / sizeof(SSplitRule));
 static void dumpLogicSubplan(const char* pRuleName, SLogicSubplan* pSubplan) {
   char* pStr = NULL;
   nodesNodeToString((SNode*)pSubplan, false, &pStr, NULL);
-  qDebugL("apply %s rule: %s", pRuleName, pStr);
+  qDebugL("apply split %s rule: %s", pRuleName, pStr);
   taosMemoryFree(pStr);
 }
 

source/libs/planner/src/planner.c

@@ -85,6 +85,11 @@ int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstream
   return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
 }
 
+int32_t qClearSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId) {
+  // todo
+  return TSDB_CODE_FAILED;
+}
+
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
   if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType && NULL == pSubplan->pNode) {
     SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;

source/libs/planner/test/planBasicTest.cpp

@@ -56,6 +56,8 @@ TEST_F(PlanBasicTest, uniqueFunc) {
 
   run("SELECT UNIQUE(c2 + 10) FROM t1 WHERE c1 > 10");
 
+  run("SELECT UNIQUE(c2 + 10), c2 FROM t1 WHERE c1 > 10");
+
   run("SELECT UNIQUE(c2 + 10), ts, c2 FROM t1 WHERE c1 > 10");
 
   run("SELECT UNIQUE(c1) a FROM t1 ORDER BY a");
@@ -75,6 +77,8 @@ TEST_F(PlanBasicTest, tailFunc) {
   run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 LIMIT 5");
 
   run("SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 PARTITION BY c1 LIMIT 5");
+
+  run("SELECT TAIL(c1, 2, 1) FROM st1s1 UNION ALL SELECT c1 FROM st1s2");
 }
 
 TEST_F(PlanBasicTest, interpFunc) {
@@ -97,6 +101,16 @@ TEST_F(PlanBasicTest, lastRowFunc) {
   run("SELECT LAST_ROW(c1) FROM st1");
 }
 
+TEST_F(PlanBasicTest, sampleFunc) {
+  useDb("root", "test");
+
+  run("SELECT SAMPLE(c1, 10) FROM t1");
+
+  run("SELECT SAMPLE(c1, 10) FROM st1");
+
+  run("SELECT SAMPLE(c1, 10) FROM st1 PARTITION BY TBNAME");
+}
+
 TEST_F(PlanBasicTest, withoutFrom) {
   useDb("root", "test");
 

source/libs/planner/test/planOptimizeTest.cpp

@@ -20,7 +20,7 @@ using namespace std;
 
 class PlanOptimizeTest : public PlannerTestBase {};
 
-TEST_F(PlanOptimizeTest, optimizeScanData) {
+TEST_F(PlanOptimizeTest, scanPath) {
   useDb("root", "test");
 
   run("SELECT COUNT(*) FROM t1");
@@ -32,7 +32,7 @@ TEST_F(PlanOptimizeTest, optimizeScanData) {
   run("SELECT PERCENTILE(c1, 40), COUNT(*) FROM t1");
 }
 
-TEST_F(PlanOptimizeTest, ConditionPushDown) {
+TEST_F(PlanOptimizeTest, pushDownCondition) {
   useDb("root", "test");
 
   run("SELECT ts, c1 FROM st1 WHERE tag1 > 4");
@@ -42,9 +42,11 @@ TEST_F(PlanOptimizeTest, ConditionPushDown) {
   run("SELECT ts, c1 FROM st1 WHERE tag1 > 4 AND tag2 = 'hello'");
 
   run("SELECT ts, c1 FROM st1 WHERE tag1 > 4 AND tag2 = 'hello' AND c1 > 10");
+
+  run("SELECT ts, c1 FROM (SELECT * FROM st1) WHERE tag1 > 4");
 }
 
-TEST_F(PlanOptimizeTest, orderByPrimaryKey) {
+TEST_F(PlanOptimizeTest, sortPrimaryKey) {
   useDb("root", "test");
 
   run("SELECT c1 FROM t1 ORDER BY ts");

source/libs/sync/inc/syncSnapshot.h

@@ -57,7 +57,7 @@ SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaI
 void                 snapshotSenderDestroy(SSyncSnapshotSender *pSender);
 bool                 snapshotSenderIsStart(SSyncSnapshotSender *pSender);
 void                 snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader);
-void                 snapshotSenderStop(SSyncSnapshotSender *pSender);
+void                 snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish);
 int32_t              snapshotSend(SSyncSnapshotSender *pSender);
 int32_t              snapshotReSend(SSyncSnapshotSender *pSender);
 
@@ -82,7 +82,7 @@ SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId from
 void                   snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
 void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg);
 bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
-void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver, bool apply);
+void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
 
 cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
 char  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);

source/libs/sync/src/syncAppendEntriesReply.c

@@ -240,7 +240,10 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
       SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId));
       ASSERT(pSender != NULL);
 
-      SSnapshot snapshot;
+      SSnapshot snapshot = {.data = NULL,
+                            .lastApplyIndex = SYNC_INDEX_INVALID,
+                            .lastApplyTerm = 0,
+                            .lastConfigIndex = SYNC_INDEX_INVALID};
       void*     pReader = NULL;
       ths->pFsm->FpGetSnapshot(ths->pFsm, &snapshot, NULL, &pReader);
       if (snapshot.lastApplyIndex >= SYNC_INDEX_BEGIN && nextIndex <= snapshot.lastApplyIndex + 1 &&
@@ -249,10 +252,6 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
         ASSERT(pReader != NULL);
         snapshotSenderStart(pSender, snapshot, pReader);
 
-        char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender start");
-        syncNodeEventLog(ths, eventLog);
-        taosMemoryFree(eventLog);
-
       } else {
         // no snapshot
         if (pReader != NULL) {
@@ -260,23 +259,6 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
         }
       }
 
-      /*
-            bool      hasSnapshot = syncNodeHasSnapshot(ths);
-            SSnapshot snapshot;
-            ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
-
-            // start sending snapshot first time
-            // start here, stop by receiver
-            if (hasSnapshot && nextIndex <= snapshot.lastApplyIndex + 1 && !snapshotSenderIsStart(pSender) &&
-                pMsg->privateTerm < pSender->privateTerm) {
-              snapshotSenderStart(pSender);
-
-              char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender start");
-              syncNodeEventLog(ths, eventLog);
-              taosMemoryFree(eventLog);
-            }
-      */
-
       SyncIndex sentryIndex = pSender->snapshot.lastApplyIndex + 1;
 
       // update nextIndex to sentryIndex
@@ -300,5 +282,5 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
   syncIndexMgrLog2("recv sync-append-entries-reply, after pNextIndex:", ths->pNextIndex);
   syncIndexMgrLog2("recv sync-append-entries-reply, after pMatchIndex:", ths->pMatchIndex);
 
-  return ret;
+  return 0;
 }
\ No newline at end of file

source/libs/sync/src/syncMain.c

@@ -698,7 +698,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
   } else {
     ret = -1;
     terrno = TSDB_CODE_SYN_NOT_LEADER;
-    sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state));
+    sError("sync propose not leader, %s", syncUtilState2String(pSyncNode->state));
     goto _END;
   }
 
@@ -1414,46 +1414,59 @@ void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
   int32_t userStrLen = strlen(str);
 
   SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
-  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
+  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
     pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
   }
-  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
-  SyncIndex logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
+
+  SyncIndex logLastIndex = SYNC_INDEX_INVALID;
+  SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
+  if (pSyncNode->pLogStore != NULL) {
+    logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
+    logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
+  }
+
+  char* pCfgStr = syncCfg2SimpleStr(&(pSyncNode->pRaftCfg->cfg));
+  char* printStr = "";
+  if (pCfgStr != NULL) {
+    printStr = pCfgStr;
+  }
 
   if (userStrLen < 256) {
-    char logBuf[128 + 256];
+    char logBuf[256 + 256];
     if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
       snprintf(logBuf, sizeof(logBuf),
                "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
                "replica-num:%d, "
-               "lconfig:%ld, changing:%d",
+               "lconfig:%ld, changing:%d, restore:%d, %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
-               pSyncNode->changing);
+               pSyncNode->changing, pSyncNode->restoreFinish, printStr);
     } else {
       snprintf(logBuf, sizeof(logBuf), "%s", str);
     }
     sError("%s", logBuf);
 
   } else {
-    int   len = 128 + userStrLen;
+    int   len = 256 + userStrLen;
     char* s = (char*)taosMemoryMalloc(len);
     if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
       snprintf(s, len,
                "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
                "replica-num:%d, "
-               "lconfig:%ld, changing:%d",
+               "lconfig:%ld, changing:%d, restore:%d, %s",
                pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
                pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
                pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
-               pSyncNode->changing);
+               pSyncNode->changing, pSyncNode->restoreFinish, printStr);
     } else {
       snprintf(s, len, "%s", str);
     }
     sError("%s", s);
     taosMemoryFree(s);
   }
+
+  taosMemoryFree(pCfgStr);
 }
 
 char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
@@ -1475,18 +1488,6 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
            pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex, pSyncNode->pRaftCfg->isStandBy,
            pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing, pSyncNode->restoreFinish);
 
-  /*
-    snprintf(s, len,
-             "syncNode: vgId:%d, term:%lu, commit:%ld, state:%d %s, standby:%d, "
-             "lc:%lu, "
-             "lc-user:%lu, "
-             "ems:%d, replica-num:%d, restore:%d, changing:%d",
-             pSyncNode->vgId, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, pSyncNode->state,
-             syncUtilState2String(pSyncNode->state), pSyncNode->pRaftCfg->isStandBy, pSyncNode->electTimerLogicClock,
-             pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS, pSyncNode->replicaNum,
-             pSyncNode->restoreFinish, pSyncNode->changing);
-  */
-
   return s;
 }
 

source/libs/sync/src/syncSnapshot.c

@@ -24,6 +24,7 @@
 //----------------------------------
 static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
                                     SyncSnapshotSend *pBeginMsg);
+static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg);
 
 //----------------------------------
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex) {
@@ -50,7 +51,7 @@ SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaI
     pSender->pSyncNode->pFsm->FpGetSnapshotInfo(pSender->pSyncNode->pFsm, &(pSender->snapshot));
     pSender->finish = false;
   } else {
-    sError("vgId:%d cannot create snapshot sender", pSyncNode->vgId);
+    sError("vgId:%d, cannot create snapshot sender", pSyncNode->vgId);
   }
 
   return pSender;
@@ -127,7 +128,7 @@ void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void
       syncEntryDestory(pEntry);
     } else {
       if (pSender->snapshot.lastConfigIndex == pSender->pSyncNode->pRaftCfg->lastConfigIndex) {
-        sTrace("vgId:%d sync sender get cfg from local", pSender->pSyncNode->vgId);
+        sTrace("vgId:%d, sync sender get cfg from local", pSender->pSyncNode->vgId);
         pSender->lastConfig = pSender->pSyncNode->pRaftCfg->cfg;
         getLastConfig = true;
       }
@@ -176,13 +177,13 @@ void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void
 
   // event log
   do {
-    char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender send");
+    char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender start");
     syncNodeEventLog(pSender->pSyncNode, eventLog);
     taosMemoryFree(eventLog);
   } while (0);
 }
 
-void snapshotSenderStop(SSyncSnapshotSender *pSender) {
+void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
   // close reader
   if (pSender->pReader != NULL) {
     int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader);
@@ -199,6 +200,14 @@ void snapshotSenderStop(SSyncSnapshotSender *pSender) {
 
   // update flag
   pSender->start = false;
+  pSender->finish = finish;
+
+  // event log
+  do {
+    char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender stop");
+    syncNodeEventLog(pSender->pSyncNode, eventLog);
+    taosMemoryFree(eventLog);
+  } while (0);
 }
 
 // when sender receive ack, call this function to send msg from seq
@@ -354,8 +363,9 @@ char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
   uint16_t port;
   syncUtilU642Addr(destId.addr, host, sizeof(host), &port);
 
-  snprintf(s, len, "%s %p laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu replica-index:%d %s:%d",
-           event, pSender, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm,
+  snprintf(s, len,
+           "%s {%p laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu replica-index:%d %s:%d}", event,
+           pSender, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm,
            pSender->snapshot.lastConfigIndex, pSender->seq, pSender->ack, pSender->finish, pSender->privateTerm,
            pSender->replicaIndex, host, port);
 
@@ -386,7 +396,7 @@ SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId from
     pReceiver->snapshot.lastConfigIndex = SYNC_INDEX_INVALID;
 
   } else {
-    sError("vgId:%d cannot create snapshot receiver", pSyncNode->vgId);
+    sError("vgId:%d, cannot create snapshot receiver", pSyncNode->vgId);
   }
 
   return pReceiver;
@@ -409,9 +419,9 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
 
 bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceiver->start; }
 
-// static do start
+// static do start by privateTerm, pBeginMsg
 // receive first snapshot data
-// privateTerm, pBeginMsg
+// write first block data
 static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
                                     SyncSnapshotSend *pBeginMsg) {
   // update state
@@ -419,6 +429,7 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p
   pReceiver->privateTerm = privateTerm;
   pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN;
   pReceiver->fromId = pBeginMsg->srcId;
+  pReceiver->start = true;
 
   // update snapshot
   pReceiver->snapshot.lastApplyIndex = pBeginMsg->lastIndex;
@@ -429,8 +440,16 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p
   ASSERT(pReceiver->pWriter == NULL);
   int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStartWrite(pReceiver->pSyncNode->pFsm, &(pReceiver->pWriter));
   ASSERT(ret == 0);
+
+  // event log
+  do {
+    char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver start");
+    syncNodeEventLog(pReceiver->pSyncNode, eventLog);
+    taosMemoryFree(eventLog);
+  } while (0);
 }
 
+// force stop
 static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
   // force close, abandon incomplete data
   if (pReceiver->pWriter != NULL) {
@@ -441,33 +460,35 @@ static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
   }
 
   pReceiver->start = false;
+
+  // event log
+  do {
+    char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver force stop");
+    syncNodeEventLog(pReceiver->pSyncNode, eventLog);
+    taosMemoryFree(eventLog);
+  } while (0);
 }
 
 // if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
 // if already start, force close, start again
 void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) {
   if (!snapshotReceiverIsStart(pReceiver)) {
-    // start
+    // first start
     snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
-    pReceiver->start = true;
 
   } else {
     // already start
-    sInfo("snapshot recv, receiver already start");
+    sInfo("vgId:%d, snapshot recv, receiver already start", pReceiver->pSyncNode->vgId);
 
     // force close, abandon incomplete data
-    int32_t ret =
-        pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false);
-    ASSERT(ret == 0);
-    pReceiver->pWriter = NULL;
+    snapshotReceiverForceStop(pReceiver);
 
     // start again
     snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
-    pReceiver->start = true;
   }
 }
 
-void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver, bool apply) {
+void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
   if (pReceiver->pWriter != NULL) {
     int32_t ret =
         pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false);
@@ -477,8 +498,69 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver, bool apply) {
 
   pReceiver->start = false;
 
-  if (apply) {
-    //    ++(pReceiver->privateTerm);
+  // event log
+  do {
+    SSnapshot snapshot;
+    pReceiver->pSyncNode->pFsm->FpGetSnapshotInfo(pReceiver->pSyncNode->pFsm, &snapshot);
+    char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver stop");
+    syncNodeEventLog(pReceiver->pSyncNode, eventLog);
+    taosMemoryFree(eventLog);
+  } while (0);
+}
+
+static void snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
+  ASSERT(pMsg->seq == SYNC_SNAPSHOT_SEQ_END);
+
+  if (pReceiver->pWriter != NULL) {
+    int32_t code = 0;
+    if (pMsg->dataLen > 0) {
+      code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data,
+                                                           pMsg->dataLen);
+      ASSERT(code == 0);
+    }
+
+    code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true);
+    ASSERT(code == 0);
+    pReceiver->pWriter = NULL;
+  }
+
+  pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
+
+  // update commit index
+  if (pReceiver->snapshot.lastApplyIndex > pReceiver->pSyncNode->commitIndex) {
+    pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
+  }
+
+  // reset wal
+  pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
+
+  // event log
+  do {
+    SSnapshot snapshot;
+    pReceiver->pSyncNode->pFsm->FpGetSnapshotInfo(pReceiver->pSyncNode->pFsm, &snapshot);
+    char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver got last data, finish, apply snapshot");
+    syncNodeEventLog(pReceiver->pSyncNode, eventLog);
+    taosMemoryFree(eventLog);
+  } while (0);
+}
+
+static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
+  ASSERT(pMsg->seq == pReceiver->ack + 1);
+
+  if (pReceiver->pWriter != NULL) {
+    if (pMsg->dataLen > 0) {
+      int32_t code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter,
+                                                                   pMsg->data, pMsg->dataLen);
+      ASSERT(code == 0);
+    }
+    pReceiver->ack = pMsg->seq;
+
+    // event log
+    do {
+      char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver receiving");
+      syncNodeEventLog(pReceiver->pSyncNode, eventLog);
+      taosMemoryFree(eventLog);
+    } while (0);
   }
 }
 
@@ -548,7 +630,7 @@ char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event)
   uint16_t port;
   syncUtilU642Addr(fromId.addr, host, sizeof(host), &port);
 
-  snprintf(s, len, "%s %p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld", event,
+  snprintf(s, len, "%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld}", event,
            pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port,
            pReceiver->snapshot.lastApplyIndex, pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex);
 
@@ -560,33 +642,20 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
   // get receiver
   SSyncSnapshotReceiver *pReceiver = pSyncNode->pNewNodeReceiver;
   bool                   needRsp = false;
-  int32_t                writeCode = 0;
 
   // state, term, seq/ack
   if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
       if (pMsg->seq == SYNC_SNAPSHOT_SEQ_BEGIN) {
-        // begin
+        // begin, no data
         snapshotReceiverStart(pReceiver, pMsg->privateTerm, pMsg);
-        pReceiver->ack = pMsg->seq;
         needRsp = true;
 
-        char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver begin");
-        syncNodeEventLog(pSyncNode, eventLog);
-        taosMemoryFree(eventLog);
-
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) {
         // end, finish FSM
-        writeCode = pSyncNode->pFsm->FpSnapshotDoWrite(pSyncNode->pFsm, pReceiver->pWriter, pMsg->data, pMsg->dataLen);
-        ASSERT(writeCode == 0);
-
-        pSyncNode->pFsm->FpSnapshotStopWrite(pSyncNode->pFsm, pReceiver->pWriter, true);
-        if (pReceiver->snapshot.lastApplyIndex > pReceiver->pSyncNode->commitIndex) {
-          pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
-        }
-
-        // pSyncNode->pLogStore->syncLogSetBeginIndex(pSyncNode->pLogStore, pMsg->lastIndex + 1);
-        pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pSyncNode->pLogStore, pMsg->lastIndex);
+        snapshotReceiverFinish(pReceiver, pMsg);
+        snapshotReceiverStop(pReceiver);
+        needRsp = true;
 
         // maybe update lastconfig
         if (pMsg->lastConfigIndex >= SYNC_INDEX_BEGIN) {
@@ -601,89 +670,83 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
           syncNodeDoConfigChange(pSyncNode, &newSyncCfg, pMsg->lastConfigIndex);
         }
 
-        SSnapshot snapshot;
-        pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
-
-        do {
-          char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver finish, apply snapshot");
-          syncNodeEventLog(pSyncNode, eventLog);
-          taosMemoryFree(eventLog);
-        } while (0);
-
-        pReceiver->pWriter = NULL;
-        snapshotReceiverStop(pReceiver, true);
-        pReceiver->ack = pMsg->seq;
-        needRsp = true;
-
-        do {
-          char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver stop");
-          syncNodeEventLog(pSyncNode, eventLog);
-          taosMemoryFree(eventLog);
-        } while (0);
-
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_FORCE_CLOSE) {
-        pSyncNode->pFsm->FpSnapshotStopWrite(pSyncNode->pFsm, pReceiver->pWriter, false);
-        snapshotReceiverStop(pReceiver, false);
+        // force close
+        snapshotReceiverForceStop(pReceiver);
         needRsp = false;
 
-        do {
-          char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver force close");
-          syncNodeEventLog(pSyncNode, eventLog);
-          taosMemoryFree(eventLog);
-        } while (0);
-
       } else if (pMsg->seq > SYNC_SNAPSHOT_SEQ_BEGIN && pMsg->seq < SYNC_SNAPSHOT_SEQ_END) {
         // transfering
         if (pMsg->seq == pReceiver->ack + 1) {
-          writeCode =
-              pSyncNode->pFsm->FpSnapshotDoWrite(pSyncNode->pFsm, pReceiver->pWriter, pMsg->data, pMsg->dataLen);
-          ASSERT(writeCode == 0);
-          pReceiver->ack = pMsg->seq;
+          snapshotReceiverGotData(pReceiver, pMsg);
         }
         needRsp = true;
 
+      } else {
+        // error log
         do {
-          char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver receiving");
-          syncNodeEventLog(pSyncNode, eventLog);
+          char logBuf[96];
+          snprintf(logBuf, sizeof(logBuf), "snapshot receiver recv error seq:%d, my ack:%d", pMsg->seq, pReceiver->ack);
+          char *eventLog = snapshotReceiver2SimpleStr(pReceiver, logBuf);
+          syncNodeErrorLog(pSyncNode, eventLog);
           taosMemoryFree(eventLog);
         } while (0);
 
-      } else {
-        ASSERT(0);
+        return -1;
       }
 
+      // send ack
       if (needRsp) {
+        // build msg
         SyncSnapshotRsp *pRspMsg = syncSnapshotRspBuild(pSyncNode->vgId);
         pRspMsg->srcId = pSyncNode->myRaftId;
         pRspMsg->destId = pMsg->srcId;
         pRspMsg->term = pSyncNode->pRaftStore->currentTerm;
         pRspMsg->lastIndex = pMsg->lastIndex;
         pRspMsg->lastTerm = pMsg->lastTerm;
-        pRspMsg->ack = pReceiver->ack;
-        pRspMsg->code = writeCode;
-        pRspMsg->privateTerm = pReceiver->privateTerm;
+        pRspMsg->ack = pReceiver->ack;  // receiver maybe already closed
+        pRspMsg->code = 0;
+        pRspMsg->privateTerm = pReceiver->privateTerm;  // receiver maybe already closed
 
+        // send msg
         SRpcMsg rpcMsg;
         syncSnapshotRsp2RpcMsg(pRspMsg, &rpcMsg);
         syncNodeSendMsgById(&(pRspMsg->destId), pSyncNode, &rpcMsg);
-
         syncSnapshotRspDestroy(pRspMsg);
       }
+    } else {
+      // error log
+      do {
+        char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver term not equal");
+        syncNodeErrorLog(pSyncNode, eventLog);
+        taosMemoryFree(eventLog);
+      } while (0);
     }
   } else {
-    syncNodeLog2("syncNodeOnSnapshotSendCb not follower", pSyncNode);
+    // error log
+    do {
+      char *eventLog = snapshotReceiver2SimpleStr(pReceiver, "snapshot receiver not follower");
+      syncNodeErrorLog(pSyncNode, eventLog);
+      taosMemoryFree(eventLog);
+    } while (0);
   }
 
   return 0;
 }
 
+static void snapshotSenderUpdateProgress(SSyncSnapshotSender *pSender, SyncSnapshotRsp *pMsg) {
+  ASSERT(pMsg->ack == pSender->seq);
+  pSender->ack = pMsg->ack;
+  ++(pSender->seq);
+}
+
 // sender receives ack, set seq = ack + 1, send msg from seq
 // if ack == SYNC_SNAPSHOT_SEQ_END, stop sender
 int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
   // if already drop replica, do not process
   if (!syncNodeInRaftGroup(pSyncNode, &(pMsg->srcId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
-    sInfo("recv SyncSnapshotRsp maybe replica already dropped");
-    return 0;
+    sError("vgId:%d, recv sync-snapshot-rsp, maybe replica already dropped", pSyncNode->vgId);
+    return -1;
   }
 
   // get sender
@@ -695,27 +758,49 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
       // receiver ack is finish, close sender
       if (pMsg->ack == SYNC_SNAPSHOT_SEQ_END) {
-        pSender->finish = true;
-        snapshotSenderStop(pSender);
+        snapshotSenderStop(pSender, true);
         return 0;
       }
 
       // send next msg
       if (pMsg->ack == pSender->seq) {
         // update sender ack
-        pSender->ack = pMsg->ack;
-        (pSender->seq)++;
+        snapshotSenderUpdateProgress(pSender, pMsg);
         snapshotSend(pSender);
 
       } else if (pMsg->ack == pSender->seq - 1) {
         snapshotReSend(pSender);
 
       } else {
-        ASSERT(0);
+        do {
+          char logBuf[96];
+          snprintf(logBuf, sizeof(logBuf), "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);
+          char *eventLog = snapshotSender2SimpleStr(pSender, logBuf);
+          syncNodeErrorLog(pSyncNode, eventLog);
+          taosMemoryFree(eventLog);
+        } while (0);
+
+        return -1;
       }
+    } else {
+      // error log
+      do {
+        char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender term not equal");
+        syncNodeErrorLog(pSyncNode, eventLog);
+        taosMemoryFree(eventLog);
+      } while (0);
+
+      return -1;
     }
   } else {
-    syncNodeLog2("syncNodeOnSnapshotRspCb not leader", pSyncNode);
+    // error log
+    do {
+      char *eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender not leader");
+      syncNodeErrorLog(pSyncNode, eventLog);
+      taosMemoryFree(eventLog);
+    } while (0);
+
+    return -1;
   }
 
   return 0;

source/util/src/terror.c

@@ -582,10 +582,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_PTR,               "Invalid tsma pointe
 TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_PARA,              "Invalid tsma parameters")
 TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_NO_INDEX_IN_CACHE,         "No tsma index in cache")
 
-
 //rsma
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_ENV,               "Invalid rsma env")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_STAT,              "Invalid rsma state")
+TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_QTASKINFO_CREATE,          "Rsma qtaskinfo creation error")
 
 //tq
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_NO_COMMITTED_OFFSET,         "No committed offset")

tests/pytest/util/cluster.py

@@ -26,7 +26,7 @@ class ClusterDnodes(TDDnodes):
 class ConfigureyCluster:
     """This will create defined number of dnodes and create a cluster.
        at the same time, it will return TDDnodes list:  dnodes, """
-    hostname= socket.gethostname()
+    hostname = socket.gethostname()
 
     def __init__(self):
         self.dnodes = []      
@@ -86,7 +86,13 @@ class ConfigureyCluster:
             time.sleep(1)
         else:
             tdLog.exit("create cluster with %d dnode but  check dnode not ready within 5s ! "%self.dnodeNums)
-            
 
+    def checkConnectStatus(self,dnodeNo,hostname=hostname):
+        dnodeNo = int(dnodeNo)
+        tdLog.info("check dnode-%d connection"%(dnodeNo+1))
+        hostname = socket.gethostname()
+        port = 6030 + dnodeNo*100
+        connectToDnode = tdCom.newcon(host=hostname,port=port)
+        return connectToDnode
 
 cluster = ConfigureyCluster()
\ No newline at end of file

tests/pytest/util/common.py

@@ -379,6 +379,11 @@ class TDCom:
         tdLog.info("cfgPath: %s" % cfgPath)
         return cfgPath
 
+    def newcon(self,host='localhost',port=6030,user='root',password='taosdata'):
+        con=taos.connect(host=host, user=user, password=password, port=port)
+        print(con)
+        return con
+
     def newcur(self,host='localhost',port=6030,user='root',password='taosdata'):
         cfgPath = self.getClientCfgPath()
         con=taos.connect(host=host, user=user, password=password, config=cfgPath, port=port)

tests/script/jenkins/basic.txt

@@ -8,21 +8,35 @@
 ./test.sh -f tsim/user/privilege_sysinfo.sim
 
 ## ---- db
-./test.sh -f tsim/db/create_all_options.sim
 ./test.sh -f tsim/db/alter_option.sim
 ./test.sh -f tsim/db/alter_replica_13.sim
-#./test.sh -f tsim/db/alter_replica_31.sim
+./test.sh -f tsim/db/alter_replica_31.sim
 ./test.sh -f tsim/db/basic1.sim
 ./test.sh -f tsim/db/basic2.sim
 ./test.sh -f tsim/db/basic3.sim
 ./test.sh -f tsim/db/basic6.sim
 ./test.sh -f tsim/db/basic7.sim
+#./test.sh -f tsim/db/commit.sim
+./test.sh -f tsim/db/create_all_options.sim
+#./test.sh -f tsim/db/delete_part.sim
+#./test.sh -f tsim/db/delete.sim
 ./test.sh -f tsim/db/error1.sim
 ./test.sh -f tsim/db/taosdlog.sim
 
 # ---- dnode
 ./test.sh -f tsim/dnode/balance_replica1.sim
-#./test.sh -f tsim/dnode/balance_replica3.sim
+./test.sh -f tsim/dnode/balance_replica3.sim
+./test.sh -f tsim/dnode/balance1.sim
+./test.sh -f tsim/dnode/balance2.sim
+./test.sh -f tsim/dnode/balance3.sim
+./test.sh -f tsim/dnode/balancex.sim
+#./test.sh -f tsim/dnode/cluster_alter.sim
+#./test.sh -f tsim/dnode/cluster_balance1.sim
+#./test.sh -f tsim/dnode/cluster_balance2.sim
+#./test.sh -f tsim/dnode/cluster_balance3.sim
+#./test.sh -f tsim/dnode/cluster_cache.sim
+#./test.sh -f tsim/dnode/cluster_flowctrl.sim
+#./test.sh -f tsim/dnode/cluster_vgroup100.sim
 ./test.sh -f tsim/dnode/create_dnode.sim
 ./test.sh -f tsim/dnode/drop_dnode_has_mnode.sim
 ./test.sh -f tsim/dnode/drop_dnode_has_qnode_snode.sim
@@ -30,11 +44,13 @@
 ./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica3.sim
 ./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica1.sim
 ./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica3.sim
+./test.sh -f tsim/dnode/offline_reason.sim
 ./test.sh -f tsim/dnode/redistribute_vgroup_replica1.sim
 ./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim
 ./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim
 ./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v2.sim
 ./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v3.sim
+./test.sh -f tsim/dnode/vnode_clean.sim
 
 # ---- insert
 ./test.sh -f tsim/insert/basic0.sim
@@ -72,7 +88,7 @@
 ./test.sh -f tsim/mnode/basic2.sim
 ./test.sh -f tsim/mnode/basic3.sim
 ./test.sh -f tsim/mnode/basic4.sim
-#./test.sh -f tsim/mnode/basic5.sim
+./test.sh -f tsim/mnode/basic5.sim
 
 # ---- show
 ./test.sh -f tsim/show/basic.sim
@@ -85,6 +101,7 @@
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
 ./test.sh -f tsim/stream/distributeInterval0.sim
+# ./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
 # ./test.sh -f tsim/stream/distributesession0.sim
 # ./test.sh -f tsim/stream/session0.sim
 ./test.sh -f tsim/stream/session1.sim
@@ -96,7 +113,6 @@
 ./test.sh -f tsim/stream/schedSnode.sim
 ./test.sh -f tsim/stream/windowClose.sim
 
-
 # ---- transaction
  ./test.sh -f tsim/trans/lossdata1.sim
  ./test.sh -f tsim/trans/create_db.sim
@@ -155,6 +171,19 @@
 # --- valgrind
 ./test.sh -f tsim/valgrind/checkError.sim -v
 
+# --- vnode
+#./test.sh -f tsim/vnode/replica3_basic.sim
+#./test.sh -f tsim/vnode/replica3_repeat.sim
+./test.sh -f tsim/vnode/replica3_vgroup.sim
+#./test.sh -f tsim/vnode/replica3_many.sim
+#./test.sh -f tsim/vnode/replica3_import.sim
+./test.sh -f tsim/vnode/stable_balance_replica1.sim
+./test.sh -f tsim/vnode/stable_dnode2_stop.sim
+./test.sh -f tsim/vnode/stable_dnode2.sim
+./test.sh -f tsim/vnode/stable_dnode3.sim
+./test.sh -f tsim/vnode/stable_replica3_dnode6.sim
+./test.sh -f tsim/vnode/stable_replica3_vnode3.sim
+
 # --- sync
 ./test.sh -f tsim/sync/3Replica1VgElect.sim
 ./test.sh -f tsim/sync/3Replica5VgElect.sim

tests/script/sh/move_dnode.sh

-#!/bin/bash
-
-echo "Executing move_dnode.sh"
-
-UNAME_BIN=`which uname`
-OS_TYPE=`$UNAME_BIN`
-
-SCRIPT_DIR=`dirname $0`
-cd $SCRIPT_DIR/../
-SCRIPT_DIR=`pwd`
-echo "SCRIPT_DIR: $SCRIPT_DIR"
-
-IN_TDINTERNAL="community"
-if [[ "$SCRIPT_DIR" == *"$IN_TDINTERNAL"* ]]; then
-  cd ../../..
-else
-  cd ../../
-fi
-
-TAOS_DIR=`pwd`
-TAOSD_DIR=`find . -name "taosd"|grep bin|head -n1`
-
-if [[ "$OS_TYPE" != "Darwin" ]]; then
-  cut_opt="--field="
-else
-  cut_opt="-f "
-fi
-
-if [[ "$TAOSD_DIR" == *"$IN_TDINTERNAL"* ]]; then
-  BIN_DIR=`find . -name "taosd"|grep bin|head -n1|cut -d '/' ${cut_opt}2,3`
-else
-  BIN_DIR=`find . -name "taosd"|grep bin|head -n1|cut -d '/' ${cut_opt}2`
-fi
-
-BUILD_DIR=$TAOS_DIR/$BIN_DIR/build
-
-SIM_DIR=$TAOS_DIR/sim
-
-NODE_DIR=$SIM_DIR/$NODE_NAME
-
-if [ -d "$SIM_DIR/$2" ];then
-  rm -rf $SIM_DIR/$2
-fi
-mv $SIM_DIR/$1 $SIM_DIR/$2
-
-if [[ $2 =~ "dnode2" ]];then
-  sed -i 's/serverPort             7100/serverPort             7200/g' $SIM_DIR/$2/cfg/taos.cfg
-  sed -i 's/dnode1/dnode2/g' $SIM_DIR/$2/cfg/taos.cfg
-  sed -i 's/7100/7200/g' $SIM_DIR/$2/data/dnode/dnodeEps.json
-elif [[ $2 =~ "dnode4" ]];then
-  sed -i 's/serverPort             7100/serverPort             7400/g' $SIM_DIR/$2/cfg/taos.cfg
-  sed -i 's/dnode1/dnode4/g' $SIM_DIR/$2/cfg/taos.cfg
-  sed -i 's/7100/7400/g' $SIM_DIR/dnode2/data/dnode/dnodeEps.json
-  sed -i 's/7100/7400/g' $SIM_DIR/dnode3/data/dnode/dnodeEps.json
-  sed -i 's/7100/7400/g' $SIM_DIR/$2/data/dnode/dnodeEps.json
-fi

tests/script/test-all.bat

@@ -19,7 +19,7 @@ for /F "usebackq tokens=*" %%i in (!caseFile!) do (
             call :GetTimeSeconds !time!
             set time1=!_timeTemp!
             echo Start at !time!
-            call !line:./test.sh=wtest.bat! > result_!a!.txt 2>error_!a!.txt
+            call !line:./test.sh=wtest.bat! > result_!a!.txt 2>error_!a!.txt || set /a errorlevel=8
             if errorlevel 1 ( call :colorEcho 0c "failed" &echo. && set /a exitNum=8 && echo %%i >>failed.txt ) else ( call :colorEcho 0a "Success" &echo. )
         )
     )