Merge remote-tracking branch 'origin/3.0' into feature/shm

include/libs/nodes/cmdnodes.h

@@ -199,6 +199,7 @@ typedef struct SIndexOptions {
 typedef struct SCreateIndexStmt {
   ENodeType type;
   EIndexType indexType;
+  bool ignoreExists;
   char indexName[TSDB_INDEX_NAME_LEN];
   char tableName[TSDB_TABLE_NAME_LEN];
   SNodeList* pCols;
@@ -207,6 +208,7 @@ typedef struct SCreateIndexStmt {
 
 typedef struct SDropIndexStmt {
   ENodeType type;
+  bool ignoreNotExists;
   char indexName[TSDB_INDEX_NAME_LEN];
   char tableName[TSDB_TABLE_NAME_LEN];
 } SDropIndexStmt;

include/libs/parser/parser.h

@@ -60,6 +60,8 @@ int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);
 
 void qDestroyQuery(SQuery* pQueryNode);
 
+int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
+
 #ifdef __cplusplus
 }
 #endif

source/client/test/clientTests.cpp

@@ -661,7 +661,7 @@ TEST(testCase, agg_query_tables) {
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   taos_free_result(pRes);
 
-  pRes = taos_query(pConn, "select k from tm0");
+  pRes = taos_query(pConn, "select count(*) from tu");
   if (taos_errno(pRes) != 0) {
     printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
     taos_free_result(pRes);

source/common/src/ttime.c

@@ -562,8 +562,7 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
 
     // not enough time range
     if (start < 0 || INT64_MAX - start > pInterval->interval - 1) {
-      end = start + pInterval->interval - 1;
-
+      end = taosTimeAdd(start, pInterval->interval, pInterval->intervalUnit, precision) - 1;
       while (end < t && ((start + pInterval->sliding) <= INT64_MAX)) {  // move forward to the correct time window
         start += pInterval->sliding;
 
@@ -587,7 +586,7 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
       start = taosTimeAdd(start, -pInterval->interval, pInterval->intervalUnit, precision);
     } else {
       // try to move current window to the left-hande-side, due to the offset effect.
-      int64_t end = start + pInterval->interval - 1;
+      int64_t end = taosTimeAdd(start, pInterval->interval, pInterval->intervalUnit, precision) - 1;
       ASSERT(end >= t);
       end = taosTimeAdd(end, -pInterval->sliding, pInterval->slidingUnit, precision);
       if (end >= t) {

source/dnode/mnode/impl/CMakeLists.txt

@@ -6,7 +6,7 @@ target_include_directories(
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 target_link_libraries(
-    mnode scheduler sdb wal transport cjson sync monitor
+    mnode scheduler sdb wal transport cjson sync monitor parser
 )
 
 if(${BUILD_TEST})

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

@@ -69,7 +69,7 @@ void mndCleanupSma(SMnode *pMnode) {}
 static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  int32_t  size = sizeof(SSmaObj) + pSma->exprLen + pSma->tagsFilterLen + TSDB_SMA_RESERVE_SIZE;
+  int32_t  size = sizeof(SSmaObj) + pSma->exprLen + pSma->tagsFilterLen + pSma->sqlLen + pSma->astLen + TSDB_SMA_RESERVE_SIZE;
   SSdbRaw *pRaw = sdbAllocRaw(SDB_SMA, TSDB_SMA_VER_NUMBER, size);
   if (pRaw == NULL) goto _OVER;
 

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

@@ -14,6 +14,7 @@
  */
 
 #include "mndStream.h"
+#include "parser.h"
 #include "mndAuth.h"
 #include "mndDb.h"
 #include "mndDnode.h"
@@ -218,28 +219,6 @@ static int32_t mndCheckCreateStreamReq(SCMCreateStreamReq *pCreate) {
   return 0;
 }
 
-static SArray *mndExtractNamesFromAst(const SNode *pAst) {
-  if (pAst->type != QUERY_NODE_SELECT_STMT) return NULL;
-
-  SArray *names = taosArrayInit(0, sizeof(void *));
-  if (names == NULL) {
-    return NULL;
-  }
-  SSelectStmt *pSelect = (SSelectStmt *)pAst;
-  SNodeList   *pNodes = pSelect->pProjectionList;
-  SListCell   *pCell = pNodes->pHead;
-  while (pCell != NULL) {
-    if (pCell->pNode->type != QUERY_NODE_FUNCTION) {
-      continue;
-    }
-    SFunctionNode *pFunction = (SFunctionNode *)pCell->pNode;
-    char          *name = strdup(pFunction->node.aliasName);
-    taosArrayPush(names, &name);
-    pCell = pCell->pNext;
-  }
-  return names;
-}
-
 static int32_t mndStreamGetPlanString(const char *ast, char **pStr) {
   if (NULL == ast) {
     return TSDB_CODE_SUCCESS;
@@ -269,18 +248,23 @@ static int32_t mndStreamGetPlanString(const char *ast, char **pStr) {
 
 int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) {
   SNode *pAst = NULL;
+#if 1 // TODO: remove debug info later
+  printf("ast = %s\n", ast); 
+#endif
   if (nodesStringToNode(ast, &pAst) < 0) {
     return -1;
   }
 #if 1
-  SArray *names = mndExtractNamesFromAst(pAst);
+  SSchemaWrapper sw = {0};
+  qExtractResultSchema(pAst, (int32_t*)&sw.nCols, &sw.pSchema);
+
   printf("|");
-  for (int i = 0; i < taosArrayGetSize(names); i++) {
-    printf(" %15s |", (char *)taosArrayGetP(names, i));
+  for (int i = 0; i < sw.nCols; i++) {
+    printf(" %15s |", (char *)sw.pSchema[i].name);
   }
   printf("\n=======================================================\n");
 
-  pStream->ColAlias = names;
+  pStream->ColAlias = NULL;
 #endif
 
   if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, &pStream->physicalPlan)) {

source/libs/executor/inc/executorimpl.h

@@ -665,6 +665,9 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
                                        SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
                                       int32_t fillType, char* fillVal, bool multigroupResult, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
+                                          int32_t numOfOutput);
+SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 
 SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
@@ -675,9 +678,7 @@ SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntim
 SOperatorInfo* createAllMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                            SExprInfo* pExpr, int32_t numOfOutput);
 SOperatorInfo* createTagScanOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput);
-SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
-                                          int32_t numOfOutput);
-SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
+
 SOperatorInfo* createMultiwaySortOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput,
                                               int32_t numOfRows, void* merger);
 SOperatorInfo* createGlobalAggregateOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,

source/libs/executor/src/executorimpl.c

@@ -674,12 +674,7 @@ static void getInitialStartTimeWindow(SInterval* pInterval, int32_t precision, T
 
     int64_t key = w->skey;
     while(key < ts) { // moving towards end
-      if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
-        key = taosTimeAdd(key, pInterval->sliding, pInterval->slidingUnit, precision);
-      } else {
-        key += pInterval->sliding;
-      }
-
+      key = taosTimeAdd(key, pInterval->sliding, pInterval->slidingUnit, precision);
       if (key >= ts) {
         break;
       }
@@ -695,12 +690,7 @@ static STimeWindow getActiveTimeWindow(SResultRowInfo * pResultRowInfo, int64_t
 
  if (pResultRowInfo->curPos == -1) {  // the first window, from the previous stored value
     getInitialStartTimeWindow(pInterval, precision, ts, &w, win->ekey, true);
-
-    if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
-      w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
-    } else {
-      w.ekey = w.skey + pInterval->interval - 1;
-    }
+    w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
   } else {
     w = getResultRow(pResultRowInfo, pResultRowInfo->curPos)->win;
   }
@@ -722,7 +712,7 @@ static STimeWindow getActiveTimeWindow(SResultRowInfo * pResultRowInfo, int64_t
       }
 
       w.skey = st;
-      w.ekey = w.skey + pInterval->interval - 1;
+      w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
     }
   }
   return w;
@@ -2208,7 +2198,7 @@ static SqlFunctionCtx* createSqlFunctionCtx_rv(SExprInfo* pExprInfo, int32_t num
   }
 
   for(int32_t i = 1; i < numOfOutput; ++i) {
-    (*rowCellInfoOffset)[i] = (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i].resDataInfo.interBufSize);
+    (*rowCellInfoOffset)[i] = (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i - 1].resDataInfo.interBufSize);
   }
 
   setCtxTagColumnInfo(pFuncCtx, numOfOutput);
@@ -2407,7 +2397,7 @@ static bool isCachedLastQuery(STaskAttr *pQueryAttr) {
 /////////////////////////////////////////////////////////////////////////////////////////////
 //todo refactor : return window
 void getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, int64_t keyFirst, int64_t keyLast, STimeWindow *win) {
-  assert(key >= keyFirst && key <= keyLast && pInterval->sliding <= pInterval->interval);
+  ASSERT(key >= keyFirst && key <= keyLast);
   win->skey = taosTimeTruncate(key, pInterval, precision);
 
   /*
@@ -2417,10 +2407,8 @@ void getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t ke
   if (keyFirst > (INT64_MAX - pInterval->interval)) {
     assert(keyLast - keyFirst < pInterval->interval);
     win->ekey = INT64_MAX;
-  } else if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
-    win->ekey = taosTimeAdd(win->skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
   } else {
-    win->ekey = win->skey + pInterval->interval - 1;
+    win->ekey = taosTimeAdd(win->skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
   }
 }
 

source/libs/function/CMakeLists.txt

 aux_source_directory(src FUNCTION_SRC)
+list(REMOVE_ITEM FUNCTION_SRC src/udfd.c)
 add_library(function STATIC ${FUNCTION_SRC})
 target_include_directories(
     function
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/function"
+    PUBLIC
+        "${CMAKE_SOURCE_DIR}/include/libs/function"
+        "${CMAKE_SOURCE_DIR}/contrib/libuv/include"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
 target_link_libraries(
     function
+    PUBLIC uv_a
     PRIVATE os util common nodes
-)
\ No newline at end of file
+)
+
+add_executable(runUdf test/runUdf.c)
+target_include_directories(
+        runUdf
+        PUBLIC
+            "${CMAKE_SOURCE_DIR}/include/libs/function"
+            "${CMAKE_SOURCE_DIR}/contrib/libuv/include"
+        PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+target_link_libraries(
+        runUdf
+        PUBLIC uv_a
+        PRIVATE os util common nodes function
+)
+
+add_library(udf1 MODULE test/udf1.c)
+target_include_directories(
+        udf1
+        PUBLIC
+            "${CMAKE_SOURCE_DIR}/include/libs/function"
+        PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
+#SET(EXECUTABLE_OUTPUT_PATH ${CMAKE_BINARY_DIR}/build/bin)
+add_executable(udfd src/udfd.c)
+target_include_directories(
+        udfd
+        PUBLIC
+            "${CMAKE_SOURCE_DIR}/include/libs/function"
+            "${CMAKE_SOURCE_DIR}/contrib/libuv/include"
+        PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
+
+target_link_libraries(
+        udfd
+        PUBLIC uv_a
+        PRIVATE os util common nodes function
+)
+

source/libs/function/inc/tudf.h

@@ -20,68 +20,116 @@
 extern "C" {
 #endif
 
-#include "os.h"
-#include "taoserror.h"
+//======================================================================================
+//begin API to taosd and qworker
+/**
+ * start udf dameon service
+ * @return error code
+ */
+int32_t startUdfService();
+
+/**
+ * stop udf dameon service
+ * @return error code
+ */
+int32_t stopUdfService();
 
 enum {
-  TSDB_UDF_FUNC_NORMAL = 0,
-  TSDB_UDF_FUNC_INIT,
-  TSDB_UDF_FUNC_FINALIZE,
-  TSDB_UDF_FUNC_MERGE,
-  TSDB_UDF_FUNC_DESTROY,
-  TSDB_UDF_FUNC_MAX_NUM
+  TSDB_UDF_TYPE_SCALAR = 0,
+  TSDB_UDF_TYPE_AGGREGATE = 1
 };
 
-typedef struct SUdfInit {
-  int32_t  maybe_null; /* 1 if function can return NULL */
-  uint32_t decimals;   /* for real functions */
-  uint64_t length;     /* For string functions */
-  char*    ptr;        /* free pointer for function data */
-  int32_t  const_item; /* 0 if result is independent of arguments */
-
-  // script like lua/javascript
-  void* script_ctx;
-  void (*destroyCtxFunc)(void* script_ctx);
-} SUdfInit;
+enum {
+  TSDB_UDF_SCRIPT_BIN_LIB = 0,
+  TSDB_UDF_SCRIPT_LUA = 1,
+};
 
 typedef struct SUdfInfo {
-  int32_t functionId;                    // system assigned function id
-  int32_t funcType;                      // scalar function or aggregate function
-  int8_t  resType;                       // result type
-  int16_t resBytes;                      // result byte
-  int32_t contLen;                       // content length
-  int32_t bufSize;                       // interbuf size
-  char*   name;                          // function name
-  void*   handle;                        // handle loaded in mem
-  void*   funcs[TSDB_UDF_FUNC_MAX_NUM];  // function ptr
-
-  // for script like lua/javascript only
-  int   isScript;
-  void* pScriptCtx;
-
-  SUdfInit init;
-  char*    content;
-  char*    path;
+  char   *udfName;        // function name
+  int32_t udfType;    // scalar function or aggregate function
+  int8_t    scriptType;
+  char *path;
+
+  int8_t  resType;     // result type
+  int16_t resBytes;    // result byte
+  int32_t bufSize;     //interbuf size
+
 } SUdfInfo;
 
+typedef void *UdfHandle;
+
+/**
+ * setup udf
+ * @param udf, in
+ * @param handle, out
+ * @return error code
+ */
+int32_t setupUdf(SUdfInfo* udf, UdfHandle *handle);
+
+
+enum {
+  TSDB_UDF_STEP_NORMAL = 0,
+  TSDB_UDF_STEP_MERGE,
+  TSDb_UDF_STEP_FINALIZE,
+  TSDB_UDF_STEP_MAX_NUM
+};
+/**
+ * call udf
+ * @param handle udf handle
+ * @param step
+ * @param state
+ * @param stateSize
+ * @param input
+ * @param newstate
+ * @param newStateSize
+ * @param output
+ * @return error code
+ */
+
+//TODO: must change the following after metadata flow and data flow between qworker and udfd is well defined
+typedef struct SUdfDataBlock {
+  char* data;
+  int32_t size;
+} SUdfDataBlock;
+
+int32_t callUdf(UdfHandle handle, int8_t step, char *state, int32_t stateSize, SUdfDataBlock input, char **newstate,
+                int32_t *newStateSize, SUdfDataBlock *output);
+
+/**
+ * tearn down udf
+ * @param handle
+ * @return
+ */
+int32_t teardownUdf(UdfHandle handle);
+
+// end API to taosd and qworker
+//=============================================================================================================================
+// TODO: Must change
+// begin API to UDF writer.
+
 // script
 
-typedef int32_t (*scriptInitFunc)(void* pCtx);
-typedef void (*scriptNormalFunc)(void* pCtx, char* data, int16_t iType, int16_t iBytes, int32_t numOfRows,
-                                 int64_t* ptList, int64_t key, char* dataOutput, char* tsOutput, int32_t* numOfOutput,
-                                 int16_t oType, int16_t oBytes);
-typedef void (*scriptFinalizeFunc)(void* pCtx, int64_t key, char* dataOutput, int32_t* numOfOutput);
-typedef void (*scriptMergeFunc)(void* pCtx, char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput);
-typedef void (*scriptDestroyFunc)(void* pCtx);
+//typedef int32_t (*scriptInitFunc)(void* pCtx);
+//typedef void (*scriptNormalFunc)(void* pCtx, char* data, int16_t iType, int16_t iBytes, int32_t numOfRows,
+//                                 int64_t* ptList, int64_t key, char* dataOutput, char* tsOutput, int32_t* numOfOutput,
+//                                 int16_t oType, int16_t oBytes);
+//typedef void (*scriptFinalizeFunc)(void* pCtx, int64_t key, char* dataOutput, int32_t* numOfOutput);
+//typedef void (*scriptMergeFunc)(void* pCtx, char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput);
+//typedef void (*scriptDestroyFunc)(void* pCtx);
 
 // dynamic lib
-typedef void (*udfNormalFunc)(char* data, int16_t itype, int16_t iBytes, int32_t numOfRows, int64_t* ts,
-                              char* dataOutput, char* interBuf, char* tsOutput, int32_t* numOfOutput, int16_t oType,
-                              int16_t oBytes, SUdfInit* buf);
-typedef int32_t (*udfInitFunc)(SUdfInit* data);
-typedef void (*udfFinalizeFunc)(char* dataOutput, char* interBuf, int32_t* numOfOutput, SUdfInit* buf);
-typedef void (*udfMergeFunc)(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf);
-typedef void (*udfDestroyFunc)(SUdfInit* buf);
+typedef int32_t (*TUdfInitFunc)();
+typedef void (*TUdfDestroyFunc)();
+
+typedef void (*TUdfFunc)(int8_t step,
+                         char *state, int32_t stateSize, SUdfDataBlock input,
+                         char **newstate, int32_t *newStateSize, SUdfDataBlock *output);
+
+//typedef void (*udfMergeFunc)(char *data, int32_t numOfRows, char *dataOutput, int32_t* numOfOutput);
+//typedef void (*udfFinalizeFunc)(char* state, int32_t stateSize, SUdfDataBlock *output);
+
+// end API to UDF writer
+//=======================================================================================================================
 
 #ifdef __cplusplus
 }

source/libs/function/inc/tudfInt.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_TUDF_INT_H
+#define TDENGINE_TUDF_INT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+//TODO replaces them with fnDebug
+//#define debugPrint(...) taosPrintLog("Function", DEBUG_INFO,  135, __VA_ARGS__)
+#define debugPrint(...) {fprintf(stderr, __VA_ARGS__);fprintf(stderr, "\n");}
+enum {
+  UDF_TASK_SETUP = 0,
+  UDF_TASK_CALL = 1,
+  UDF_TASK_TEARDOWN = 2
+
+};
+
+typedef struct SUdfSetupRequest {
+  char udfName[16]; //
+  int8_t scriptType; // 0:c, 1: lua, 2:js
+  int8_t udfType; //udaf, udf
+  int16_t pathSize;
+  char *path;
+} SUdfSetupRequest;
+
+typedef struct SUdfSetupResponse {
+  int64_t udfHandle;
+} SUdfSetupResponse;
+
+
+typedef struct SUdfCallRequest {
+  int64_t udfHandle;
+  int8_t step;
+
+  int32_t inputBytes;
+  char *input;
+
+  int32_t stateBytes;
+  char *state;
+} SUdfCallRequest;
+
+
+typedef struct SUdfCallResponse {
+  int32_t outputBytes;
+  char *output;
+  int32_t newStateBytes;
+  char *newState;
+} SUdfCallResponse;
+
+
+typedef struct SUdfTeardownRequest {
+  int64_t udfHandle;
+} SUdfTeardownRequest;
+
+
+typedef struct SUdfTeardownResponse {
+} SUdfTeardownResponse;
+
+typedef struct SUdfRequest {
+  int32_t msgLen;
+  int64_t seqNum;
+
+  int8_t type;
+  void *subReq;
+} SUdfRequest;
+
+typedef struct SUdfResponse {
+  int32_t msgLen;
+  int64_t seqNum;
+
+  int8_t type;
+  int32_t code;
+  void *subRsp;
+} SUdfResponse;
+
+int32_t decodeRequest(char *buf, int32_t bufLen, SUdfRequest **pRequest);
+int32_t encodeResponse(char **buf, int32_t *bufLen, SUdfResponse *response);
+int32_t encodeRequest(char **buf, int32_t *bufLen, SUdfRequest *request);
+int32_t decodeResponse(char *buf, int32_t bufLen, SUdfResponse **pResponse);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_TUDF_INT_H

source/libs/function/inc/udfc.h

+//
+// Created by shenglian on 28/02/22.
+//
+
+#ifndef UDF_UDF_H
+#define UDF_UDF_H
+
+#include <stdlib.h>
+#define DEBUG
+#ifdef DEBUG
+#define debugPrint(...) fprintf(__VA_ARGS__)
+#else
+#define debugPrint(...) /**/
+#endif
+
+enum {
+    UDF_TASK_SETUP = 0,
+    UDF_TASK_CALL = 1,
+    UDF_TASK_TEARDOWN = 2
+
+};
+
+typedef struct SSDataBlock{
+    char *data;
+    int32_t size;
+} SSDataBlock;
+
+typedef struct SUdfInfo {
+    char *udfName;
+    char *path;
+} SUdfInfo;
+
+typedef void *UdfHandle;
+
+int32_t startUdfService();
+
+int32_t stopUdfService();
+
+//int32_t setupUdf(SUdfInfo *udf, int32_t numOfUdfs, UdfHandle *handles);
+
+int32_t setupUdf(SUdfInfo* udf, UdfHandle* handle);
+
+int32_t callUdf(UdfHandle handle, int8_t step, char *state, int32_t stateSize, SSDataBlock input, char **newstate,
+                int32_t *newStateSize, SSDataBlock *output);
+
+int32_t teardownUdf(UdfHandle handle);
+
+typedef struct SUdfSetupRequest {
+    char udfName[16]; //
+    int8_t scriptType; // 0:c, 1: lua, 2:js
+    int8_t udfType; //udaf, udf, udtf
+    int16_t pathSize;
+    char *path;
+} SUdfSetupRequest;
+
+typedef struct SUdfSetupResponse {
+    int64_t udfHandle;
+} SUdfSetupResponse;
+
+
+typedef struct SUdfCallRequest {
+    int64_t udfHandle;
+    int8_t step;
+
+    int32_t inputBytes;
+    char *input;
+
+    int32_t stateBytes;
+    char *state;
+} SUdfCallRequest;
+
+
+typedef struct SUdfCallResponse {
+    int32_t outputBytes;
+    char *output;
+    int32_t newStateBytes;
+    char *newState;
+} SUdfCallResponse;
+
+
+typedef struct SUdfTeardownRequest {
+    int64_t udfHandle;
+} SUdfTeardownRequest;
+
+
+typedef struct SUdfTeardownResponse {
+} SUdfTeardownResponse;
+
+typedef struct SUdfRequest {
+    int32_t msgLen;
+    int64_t seqNum;
+
+    int8_t type;
+    void *subReq;
+} SUdfRequest;
+
+typedef struct SUdfResponse {
+    int32_t msgLen;
+    int64_t seqNum;
+
+    int8_t type;
+    int32_t code;
+    void *subRsp;
+} SUdfResponse;
+
+int32_t decodeRequest(char *buf, int32_t bufLen, SUdfRequest **pRequest);
+int32_t encodeResponse(char **buf, int32_t *bufLen, SUdfResponse *response);
+int32_t encodeRequest(char **buf, int32_t *bufLen, SUdfRequest *request);
+int32_t decodeResponse(char *buf, int32_t bufLen, SUdfResponse **pResponse);
+#endif //UDF_UDF_H

source/libs/function/src/builtinsimpl.c

@@ -528,11 +528,10 @@ void firstFunction(SqlFunctionCtx *pCtx) {
   char* buf = GET_ROWCELL_INTERBUF(pResInfo);
 
   SInputColumnInfoData* pInput = &pCtx->input;
-
   SColumnInfoData* pInputCol = pInput->pData[0];
 
   // All null data column, return directly.
-  if (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows) {
+  if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
     ASSERT(pInputCol->hasNull == true);
     return;
   }

source/libs/function/src/udfd.c

+/*
+ * 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/>.
+ */
+
+#include "uv.h"
+#include "os.h"
+#include "tlog.h"
+
+#include "tudf.h"
+#include "tudfInt.h"
+
+
+static uv_loop_t *loop;
+
+typedef struct SUdfdUvConn {
+    uv_stream_t *client;
+    char *inputBuf;
+    int32_t inputLen;
+    int32_t inputCap;
+    int32_t inputTotal;
+} SUdfdUvConn;
+
+typedef struct SUvUdfWork {
+    uv_stream_t *client;
+    uv_buf_t input;
+    uv_buf_t output;
+} SUvUdfWork;
+
+typedef struct SUdf {
+    int32_t refCount;
+
+    char name[16];
+    int8_t type;
+
+    uv_lib_t lib;
+    TUdfFunc normalFunc;
+} SUdf;
+
+//TODO: low priority: change name onxxx to xxxCb, and udfc or udfd as prefix
+//TODO: add private udf structure.
+typedef struct SUdfHandle {
+    SUdf *udf;
+} SUdfHandle;
+
+
+void udfdProcessRequest(uv_work_t *req) {
+    SUvUdfWork *uvUdf = (SUvUdfWork *) (req->data);
+    SUdfRequest *request = NULL;
+    decodeRequest(uvUdf->input.base, uvUdf->input.len, &request);
+
+    switch (request->type) {
+        case UDF_TASK_SETUP: {
+            debugPrint("%s", "process setup request");
+            SUdf *udf = malloc(sizeof(SUdf));
+            udf->refCount = 0;
+            SUdfSetupRequest *setup = request->subReq;
+            strcpy(udf->name, setup->udfName);
+            int err = uv_dlopen(setup->path, &udf->lib);
+            if (err != 0) {
+                debugPrint("can not load library %s. error: %s", setup->path, uv_strerror(err));
+                //TODO set error
+            }
+
+            char normalFuncName[32] = {0};
+            strcpy(normalFuncName, setup->udfName);
+	    //TODO error,
+	    //TODO find all functions normal, init, destroy, normal, merge, finalize
+            uv_dlsym(&udf->lib, normalFuncName, (void **) (&udf->normalFunc));
+
+            SUdfHandle *handle = malloc(sizeof(SUdfHandle));
+            handle->udf = udf;
+            udf->refCount++;
+            //TODO: allocate private structure and call init function and set it to handle
+            SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
+            rsp->seqNum = request->seqNum;
+            rsp->type = request->type;
+            rsp->code = 0;
+            SUdfSetupResponse *subRsp = malloc(sizeof(SUdfSetupResponse));
+            subRsp->udfHandle = (int64_t) (handle);
+            rsp->subRsp = subRsp;
+            char *buf;
+            int32_t len;
+            encodeResponse(&buf, &len, rsp);
+
+            uvUdf->output = uv_buf_init(buf, len);
+
+            free(rsp->subRsp);
+            free(rsp);
+            free(request->subReq);
+            free(request);
+            free(uvUdf->input.base);
+            break;
+        }
+
+        case UDF_TASK_CALL: {
+            debugPrint("%s", "process call request");
+            SUdfCallRequest *call = request->subReq;
+            SUdfHandle *handle = (SUdfHandle *) (call->udfHandle);
+            SUdf *udf = handle->udf;
+            char *newState;
+            int32_t newStateSize;
+            SUdfDataBlock input = {.data = call->input, .size= call->inputBytes};
+            SUdfDataBlock output;
+	    //TODO: call different functions according to the step 
+            udf->normalFunc(call->step, call->state, call->stateBytes, input, &newState, &newStateSize, &output);
+
+            SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
+            rsp->seqNum = request->seqNum;
+            rsp->type = request->type;
+            rsp->code = 0;
+            SUdfCallResponse *subRsp = malloc(sizeof(SUdfCallResponse));
+            subRsp->outputBytes = output.size;
+            subRsp->output = output.data;
+            subRsp->newStateBytes = newStateSize;
+            subRsp->newState = newState;
+            rsp->subRsp = subRsp;
+
+            char *buf;
+            int32_t len;
+            encodeResponse(&buf, &len, rsp);
+            uvUdf->output = uv_buf_init(buf, len);
+
+            free(rsp->subRsp);
+            free(rsp);
+            free(newState);
+            free(output.data);
+            free(request->subReq);
+            free(request);
+            free(uvUdf->input.base);
+            break;
+        }
+        case UDF_TASK_TEARDOWN: {
+            debugPrint("%s", "process teardown request");
+
+            SUdfTeardownRequest *teardown = request->subReq;
+            SUdfHandle *handle = (SUdfHandle *) (teardown->udfHandle);
+            SUdf *udf = handle->udf;
+            udf->refCount--;
+            if (udf->refCount == 0) {
+                uv_dlclose(&udf->lib);
+            }
+            free(udf);
+	    //TODO: call destroy and free udf private 
+            free(handle);
+
+            SUdfResponse *rsp = malloc(sizeof(SUdfResponse));
+            rsp->seqNum = request->seqNum;
+            rsp->type = request->type;
+            rsp->code = 0;
+            SUdfTeardownResponse *subRsp = malloc(sizeof(SUdfTeardownResponse));
+            rsp->subRsp = subRsp;
+            char *buf;
+            int32_t len;
+            encodeResponse(&buf, &len, rsp);
+            uvUdf->output = uv_buf_init(buf, len);
+
+            free(rsp->subRsp);
+            free(rsp);
+            free(request->subReq);
+            free(request);
+            free(uvUdf->input.base);
+            break;
+        }
+        default: {
+            break;
+        }
+
+    }
+
+}
+
+void udfdOnWrite(uv_write_t *req, int status) {
+    debugPrint("%s", "after writing to pipe");
+    if (status < 0) {
+        debugPrint("Write error %s", uv_err_name(status));
+    }
+    SUvUdfWork *work = (SUvUdfWork *) req->data;
+    debugPrint("\tlength: %zu", work->output.len);
+    free(work->output.base);
+    free(work);
+    free(req);
+}
+
+
+void udfdSendResponse(uv_work_t *work, int status) {
+    debugPrint("%s", "send response");
+    SUvUdfWork *udfWork = (SUvUdfWork *) (work->data);
+
+    uv_write_t *write_req = malloc(sizeof(uv_write_t));
+    write_req->data = udfWork;
+    uv_write(write_req, udfWork->client, &udfWork->output, 1, udfdOnWrite);
+
+    free(work);
+}
+
+void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
+    debugPrint("%s", "allocate buffer for read");
+    SUdfdUvConn *ctx = handle->data;
+    int32_t msgHeadSize = sizeof(int32_t) + sizeof(int64_t);
+    if (ctx->inputCap == 0) {
+        ctx->inputBuf = malloc(msgHeadSize);
+        if (ctx->inputBuf) {
+            ctx->inputLen = 0;
+            ctx->inputCap = msgHeadSize;
+            ctx->inputTotal = -1;
+
+            buf->base = ctx->inputBuf;
+            buf->len = ctx->inputCap;
+        } else {
+            //TODO: log error
+            buf->base = NULL;
+            buf->len = 0;
+        }
+    } else {
+        ctx->inputCap = ctx->inputTotal > ctx->inputCap ? ctx->inputTotal : ctx->inputCap;
+        void *inputBuf = realloc(ctx->inputBuf, ctx->inputCap);
+        if (inputBuf) {
+            ctx->inputBuf = inputBuf;
+            buf->base = ctx->inputBuf + ctx->inputLen;
+            buf->len = ctx->inputCap - ctx->inputLen;
+        } else {
+            //TODO: log error
+            buf->base = NULL;
+            buf->len = 0;
+        }
+    }
+    debugPrint("\tinput buf cap - len - total : %d - %d - %d", ctx->inputCap, ctx->inputLen, ctx->inputTotal);
+
+}
+
+bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
+    if (pipe->inputTotal == -1 && pipe->inputLen >= sizeof(int32_t)) {
+        pipe->inputTotal = *(int32_t *) (pipe->inputBuf);
+    }
+    if (pipe->inputLen == pipe->inputCap && pipe->inputTotal == pipe->inputCap) {
+        return true;
+    }
+    return false;
+}
+
+void udfdHandleRequest(SUdfdUvConn *conn) {
+    uv_work_t *work = malloc(sizeof(uv_work_t));
+    SUvUdfWork *udfWork = malloc(sizeof(SUvUdfWork));
+    udfWork->client = conn->client;
+    udfWork->input = uv_buf_init(conn->inputBuf, conn->inputLen);
+    conn->inputBuf = NULL;
+    conn->inputLen = 0;
+    conn->inputCap = 0;
+    conn->inputTotal = -1;
+    work->data = udfWork;
+    uv_queue_work(loop, work, udfdProcessRequest, udfdSendResponse);
+}
+
+void udfdPipeCloseCb(uv_handle_t *pipe) {
+    SUdfdUvConn *conn = pipe->data;
+    free(conn->client);
+    free(conn->inputBuf);
+    free(conn);
+}
+
+void udfdUvHandleError(SUdfdUvConn *conn) {
+    uv_close((uv_handle_t *) conn->client, udfdPipeCloseCb);
+}
+
+void udfdPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
+    debugPrint("%s, nread: %zd", "read from pipe", nread);
+
+    if (nread == 0) return;
+
+    SUdfdUvConn *conn = client->data;
+
+    if (nread > 0) {
+        conn->inputLen += nread;
+        if (isUdfdUvMsgComplete(conn)) {
+            udfdHandleRequest(conn);
+        } else {
+            //log error or continue;
+        }
+        return;
+    }
+
+    if (nread < 0) {
+        debugPrint("Read error %s", uv_err_name(nread));
+        if (nread == UV_EOF) {
+            //TODO check more when close
+        } else {
+        }
+        udfdUvHandleError(conn);
+    }
+}
+
+void udfdOnNewConnection(uv_stream_t *server, int status) {
+    debugPrint("%s", "on new connection");
+    if (status < 0) {
+        // TODO
+        return;
+    }
+
+    uv_pipe_t *client = (uv_pipe_t *) malloc(sizeof(uv_pipe_t));
+    uv_pipe_init(loop, client, 0);
+    if (uv_accept(server, (uv_stream_t *) client) == 0) {
+        SUdfdUvConn *ctx = malloc(sizeof(SUdfdUvConn));
+        ctx->client = (uv_stream_t *) client;
+        ctx->inputBuf = 0;
+        ctx->inputLen = 0;
+        ctx->inputCap = 0;
+        client->data = ctx;
+        ctx->client = (uv_stream_t *) client;
+        uv_read_start((uv_stream_t *) client, udfdAllocBuffer, udfdPipeRead);
+    } else {
+        uv_close((uv_handle_t *) client, NULL);
+    }
+}
+
+void removeListeningPipe(int sig) {
+    uv_fs_t req;
+    uv_fs_unlink(loop, &req, "udf.sock", NULL);
+    exit(0);
+}
+
+int main() {
+    debugPrint("libuv version: %x", UV_VERSION_HEX);
+
+    loop = uv_default_loop();
+    uv_fs_t req;
+    uv_fs_unlink(loop, &req, "udf.sock", NULL);
+
+    uv_pipe_t server;
+    uv_pipe_init(loop, &server, 0);
+
+    signal(SIGINT, removeListeningPipe);
+
+    int r;
+    if ((r = uv_pipe_bind(&server, "udf.sock"))) {
+        debugPrint("Bind error %s\n", uv_err_name(r));
+        removeListeningPipe(0);
+        return 1;
+    }
+    if ((r = uv_listen((uv_stream_t *) &server, 128, udfdOnNewConnection))) {
+        debugPrint("Listen error %s", uv_err_name(r));
+        return 2;
+    }
+    uv_run(loop, UV_RUN_DEFAULT);
+    uv_loop_close(loop);
+}

source/libs/function/test/runUdf.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "uv.h"
+#include "tudf.h"
+
+int main(int argc, char *argv[]) {
+    startUdfService();
+    uv_sleep(1000);
+    char path[256] = {0};
+    size_t cwdSize = 256;
+    int err = uv_cwd(path, &cwdSize);
+    if (err != 0) {
+        fprintf(stderr, "err cwd: %s\n", uv_strerror(err));
+	    return err;
+    }
+    fprintf(stdout, "current working directory:%s\n", path);
+    strcat(path, "/libudf1.so");
+    SUdfInfo udfInfo = {.udfName="udf1", .path=path};
+
+    UdfHandle handle;
+    setupUdf(&udfInfo, &handle);
+
+    //char state[5000000] = "state";
+    //char input[5000000] = "input";
+    int dataSize = 500;
+    int callCount = 2;
+    if (argc > 1) dataSize = atoi(argv[1]);
+    if (argc > 2) callCount = atoi(argv[2]);
+    char *state = malloc(dataSize);
+    char *input = malloc(dataSize);
+    SUdfDataBlock blockInput = {.data = input, .size = dataSize};
+    SUdfDataBlock blockOutput;
+    char* newState;
+    int32_t newStateSize;
+    for (int l = 0; l < callCount; ++l) {
+        callUdf(handle, 0, state, dataSize, blockInput, &newState, &newStateSize, &blockOutput);
+    }
+    free(state);
+    free(input);
+    teardownUdf(handle);
+
+    stopUdfService();
+}

source/libs/function/test/udf1.c

+#include <string.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#include "tudf.h"
+
+void udf1(int8_t step, char *state, int32_t stateSize, SUdfDataBlock input,
+          char **newState, int32_t *newStateSize, SUdfDataBlock *output) {
+    fprintf(stdout, "%s, step:%d\n", "udf function called", step);
+    char *newStateBuf = malloc(stateSize);
+    memcpy(newStateBuf, state, stateSize);
+    *newState = newStateBuf;
+    *newStateSize = stateSize;
+
+    char *outputBuf = malloc(input.size);
+    memcpy(outputBuf, input.data, input.size);
+    output->data = outputBuf;
+    output->size = input.size;
+    return;
+}

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1480,7 +1480,7 @@ static int32_t jsonToDatum(const SJson* pJson, void* pObj) {
     case TSDB_DATA_TYPE_NCHAR:
     case TSDB_DATA_TYPE_VARCHAR:
     case TSDB_DATA_TYPE_VARBINARY: {
-      pNode->datum.p = calloc(1, pNode->node.resType.bytes);
+      pNode->datum.p = calloc(1, pNode->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
       if (NULL == pNode->datum.p) {
         code = TSDB_CODE_OUT_OF_MEMORY;
         break;

source/libs/parser/inc/parAst.h

@@ -150,9 +150,9 @@ SNode* createDropUserStmt(SAstCreateContext* pCxt, SToken* pUserName);
 SNode* createCreateDnodeStmt(SAstCreateContext* pCxt, const SToken* pFqdn, const SToken* pPort);
 SNode* createDropDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode);
 SNode* createAlterDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode, const SToken* pConfig, const SToken* pValue);
-SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, SToken* pIndexName, SToken* pTableName, SNodeList* pCols, SNode* pOptions);
+SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, bool ignoreExists, SToken* pIndexName, SToken* pTableName, SNodeList* pCols, SNode* pOptions);
 SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInterval, SNode* pOffset, SNode* pSliding);
-SNode* createDropIndexStmt(SAstCreateContext* pCxt, SToken* pIndexName, SToken* pTableName);
+SNode* createDropIndexStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToken* pIndexName, SToken* pTableName);
 SNode* createCreateQnodeStmt(SAstCreateContext* pCxt, const SToken* pDnodeId);
 SNode* createDropQnodeStmt(SAstCreateContext* pCxt, const SToken* pDnodeId);
 SNode* createCreateTopicStmt(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, SNode* pQuery, const SToken* pSubscribeDbName);

source/libs/parser/inc/sql.y

@@ -313,10 +313,11 @@ func_name_list(A) ::= func_name_list(B) NK_COMMA col_name(C).
 func_name(A) ::= function_name(B).                                                { A = createFunctionNode(pCxt, &B, NULL); }
 
 /************************************************ create index ********************************************************/
-cmd ::= CREATE SMA INDEX index_name(A) ON table_name(B) index_options(C).         { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_SMA, &A, &B, NULL, C); }
-cmd ::= CREATE FULLTEXT INDEX
-  index_name(A) ON table_name(B) NK_LP col_name_list(C) NK_RP.                    { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_FULLTEXT, &A, &B, C, NULL); }
-cmd ::= DROP INDEX index_name(A) ON table_name(B).                                { pCxt->pRootNode = createDropIndexStmt(pCxt, &A, &B); }
+cmd ::= CREATE SMA INDEX not_exists_opt(D) 
+  index_name(A) ON table_name(B) index_options(C).                                { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_SMA, D, &A, &B, NULL, C); }
+cmd ::= CREATE FULLTEXT INDEX not_exists_opt(D)
+  index_name(A) ON table_name(B) NK_LP col_name_list(C) NK_RP.                    { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_FULLTEXT, D, &A, &B, C, NULL); }
+cmd ::= DROP INDEX exists_opt(C) index_name(A) ON table_name(B).                  { pCxt->pRootNode = createDropIndexStmt(pCxt, C, &A, &B); }
 
 index_options(A) ::= .                                                            { A = NULL; }
 index_options(A) ::= FUNCTION NK_LP func_list(B) NK_RP INTERVAL 

source/libs/parser/src/parAstCreater.c

@@ -1160,13 +1160,14 @@ SNode* createAlterDnodeStmt(SAstCreateContext* pCxt, const SToken* pDnode, const
   return (SNode*)pStmt;
 }
 
-SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, SToken* pIndexName, SToken* pTableName, SNodeList* pCols, SNode* pOptions) {
+SNode* createCreateIndexStmt(SAstCreateContext* pCxt, EIndexType type, bool ignoreExists, SToken* pIndexName, SToken* pTableName, SNodeList* pCols, SNode* pOptions) {
   if (!checkIndexName(pCxt, pIndexName) || !checkTableName(pCxt, pTableName)) {
     return NULL;
   }
   SCreateIndexStmt* pStmt = nodesMakeNode(QUERY_NODE_CREATE_INDEX_STMT);
   CHECK_OUT_OF_MEM(pStmt);
   pStmt->indexType = type;
+  pStmt->ignoreExists = ignoreExists;
   strncpy(pStmt->indexName, pIndexName->z, pIndexName->n);
   strncpy(pStmt->tableName, pTableName->z, pTableName->n);
   pStmt->pCols = pCols;
@@ -1184,12 +1185,13 @@ SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInt
   return (SNode*)pOptions;
 }
 
-SNode* createDropIndexStmt(SAstCreateContext* pCxt, SToken* pIndexName, SToken* pTableName) {
+SNode* createDropIndexStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SToken* pIndexName, SToken* pTableName) {
   if (!checkIndexName(pCxt, pIndexName) || !checkTableName(pCxt, pTableName)) {
     return NULL;
   }
   SDropIndexStmt* pStmt = nodesMakeNode(QUERY_NODE_DROP_INDEX_STMT);
   CHECK_OUT_OF_MEM(pStmt);
+  pStmt->ignoreNotExists = ignoreNotExists;
   strncpy(pStmt->indexName, pIndexName->z, pIndexName->n);
   strncpy(pStmt->tableName, pTableName->z, pTableName->n);
   return (SNode*)pStmt;

source/libs/parser/src/parTranslater.c

@@ -383,7 +383,7 @@ static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal) {
       case TSDB_DATA_TYPE_NCHAR:
       case TSDB_DATA_TYPE_VARCHAR:
       case TSDB_DATA_TYPE_VARBINARY: {
-        pVal->datum.p = calloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE);
+        pVal->datum.p = calloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
         if (NULL == pVal->datum.p) {
           return generateDealNodeErrMsg(pCxt, TSDB_CODE_OUT_OF_MEMORY);
         }
@@ -1406,34 +1406,110 @@ static int32_t translateShowTables(STranslateContext* pCxt) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt* pStmt) {
-  SVCreateTSmaReq createSmaReq = {0};
+static int32_t getSmaIndexDstVgId(STranslateContext* pCxt, char* pTableName, int32_t* pVgId) {
+  SVgroupInfo vg = {0};
+  int32_t code = getTableHashVgroup(pCxt, pCxt->pParseCxt->db, pTableName, &vg);
+  if (TSDB_CODE_SUCCESS == code) {
+    *pVgId = vg.vgId;
+  }
+  return code;
+}
 
-  if (DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pInterval) ||
-      (NULL != pStmt->pOptions->pOffset && DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pOffset)) ||
-      (NULL != pStmt->pOptions->pSliding && DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pSliding))) {
-    return pCxt->errCode;
+static int32_t getSmaIndexSql(STranslateContext* pCxt, char** pSql, int32_t* pLen) {
+  *pSql = strdup(pCxt->pParseCxt->pSql);
+  if (NULL == *pSql) {
+    return TSDB_CODE_OUT_OF_MEMORY;
   }
+  *pLen = pCxt->pParseCxt->sqlLen + 1;
+  return TSDB_CODE_SUCCESS;
+}
 
-  createSmaReq.tSma.intervalUnit = ((SValueNode*)pStmt->pOptions->pInterval)->unit;
-  createSmaReq.tSma.slidingUnit = (NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->unit : 0);
-  strcpy(createSmaReq.tSma.indexName, pStmt->indexName);
+static int32_t getSmaIndexExpr(STranslateContext* pCxt, SCreateIndexStmt* pStmt, char** pExpr, int32_t* pLen) {
+  return nodesListToString(pStmt->pOptions->pFuncs, false, pExpr, pLen);
+}
 
-  SName name;
-  name.type = TSDB_TABLE_NAME_T;
-  name.acctId = pCxt->pParseCxt->acctId;
+static int32_t getSmaIndexBuildAst(STranslateContext* pCxt, SCreateIndexStmt* pStmt, char** pAst, int32_t* pLen) {
+  SSelectStmt* pSelect = nodesMakeNode(QUERY_NODE_SELECT_STMT);
+  if (NULL == pSelect) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  SRealTableNode* pTable = nodesMakeNode(QUERY_NODE_REAL_TABLE);
+  if (NULL == pTable) {
+    nodesDestroyNode(pSelect);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  strcpy(pTable->table.dbName, pCxt->pParseCxt->db);
+  strcpy(pTable->table.tableName, pStmt->tableName);
+  pSelect->pFromTable = (SNode*)pTable;
+
+  pSelect->pProjectionList = nodesCloneList(pStmt->pOptions->pFuncs);
+  if (NULL == pTable) {
+    nodesDestroyNode(pSelect);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  SIntervalWindowNode* pInterval = nodesMakeNode(QUERY_NODE_INTERVAL_WINDOW);
+  if (NULL == pInterval) {
+    nodesDestroyNode(pSelect);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  pSelect->pWindow = (SNode*)pInterval;
+  pInterval->pInterval = nodesCloneNode(pStmt->pOptions->pInterval);
+  pInterval->pOffset = nodesCloneNode(pStmt->pOptions->pOffset);
+  pInterval->pSliding = nodesCloneNode(pStmt->pOptions->pSliding);
+  if (NULL == pInterval->pInterval || (NULL != pStmt->pOptions->pOffset && NULL == pInterval->pOffset) ||
+      (NULL != pStmt->pOptions->pSliding && NULL == pInterval->pSliding)) {
+    nodesDestroyNode(pSelect);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  int32_t code = translateQuery(pCxt, (SNode*)pSelect);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = nodesNodeToString(pSelect, false, pAst, pLen);
+  }
+  nodesDestroyNode(pSelect);
+  return code;
+}
+
+static int32_t buildCreateSmaReq(STranslateContext* pCxt, SCreateIndexStmt* pStmt, SMCreateSmaReq* pReq) {
+  SName name = { .type = TSDB_TABLE_NAME_T, .acctId = pCxt->pParseCxt->acctId };
   strcpy(name.dbname, pCxt->pParseCxt->db);
+  strcpy(name.tname, pStmt->indexName);
+  tNameExtractFullName(&name, pReq->name);
   strcpy(name.tname, pStmt->tableName);
-  STableMeta* pMeta = NULL;
-  int32_t code = catalogGetTableMeta(pCxt->pParseCxt->pCatalog, pCxt->pParseCxt->pTransporter, &pCxt->pParseCxt->mgmtEpSet, &name, &pMeta);
-  if (TSDB_CODE_SUCCESS != code) {
-    return code;
+  name.tname[strlen(pStmt->tableName)] = '\0';
+  tNameExtractFullName(&name, pReq->stb);
+  pReq->igExists = pStmt->ignoreExists;
+  pReq->interval = ((SValueNode*)pStmt->pOptions->pInterval)->datum.i;
+  pReq->intervalUnit = ((SValueNode*)pStmt->pOptions->pInterval)->unit;
+  pReq->offset = (NULL != pStmt->pOptions->pOffset ? ((SValueNode*)pStmt->pOptions->pOffset)->datum.i : 0);
+  pReq->sliding = (NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->datum.i : pReq->interval);
+  pReq->slidingUnit = (NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->unit : pReq->intervalUnit);
+
+  int32_t code = getSmaIndexDstVgId(pCxt, pStmt->tableName, &pReq->dstVgId);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = getSmaIndexSql(pCxt, &pReq->sql, &pReq->sqlLen);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = getSmaIndexExpr(pCxt, pStmt, &pReq->expr, &pReq->exprLen);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = getSmaIndexBuildAst(pCxt, pStmt, &pReq->ast, &pReq->astLen);
+  }
+
+  return code;
+}
+
+static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt* pStmt) {
+  if (DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pInterval) ||
+      (NULL != pStmt->pOptions->pOffset && DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pOffset)) ||
+      (NULL != pStmt->pOptions->pSliding && DEAL_RES_ERROR == translateValue(pCxt, (SValueNode*)pStmt->pOptions->pSliding))) {
+    return pCxt->errCode;
   }
 
-  createSmaReq.tSma.tableUid = pMeta->uid;
-  createSmaReq.tSma.interval = ((SValueNode*)pStmt->pOptions->pInterval)->datum.i;
-  createSmaReq.tSma.sliding = (NULL != pStmt->pOptions->pSliding ? ((SValueNode*)pStmt->pOptions->pSliding)->datum.i : 0);
-  code = nodesListToString(pStmt->pOptions->pFuncs, false, &createSmaReq.tSma.expr, &createSmaReq.tSma.exprLen);
+  SMCreateSmaReq createSmaReq = {0};
+  int32_t code = buildCreateSmaReq(pCxt, pStmt, &createSmaReq);
   if (TSDB_CODE_SUCCESS != code) {
     return code;
   }
@@ -1443,15 +1519,14 @@ static int32_t translateCreateSmaIndex(STranslateContext* pCxt, SCreateIndexStmt
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   pCxt->pCmdMsg->epSet = pCxt->pParseCxt->mgmtEpSet;
-  pCxt->pCmdMsg->msgType = TDMT_VND_CREATE_SMA;
-  pCxt->pCmdMsg->msgLen = tSerializeSVCreateTSmaReq(NULL, &createSmaReq);
+  pCxt->pCmdMsg->msgType = TDMT_MND_CREATE_SMA;
+  pCxt->pCmdMsg->msgLen = tSerializeSMCreateSmaReq(NULL, 0, &createSmaReq);
   pCxt->pCmdMsg->pMsg = malloc(pCxt->pCmdMsg->msgLen);
   if (NULL == pCxt->pCmdMsg->pMsg) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
-  void* pBuf = pCxt->pCmdMsg->pMsg;
-  tSerializeSVCreateTSmaReq(&pBuf, &createSmaReq);
-  tdDestroyTSma(&createSmaReq.tSma);
+  tSerializeSMCreateSmaReq(pCxt->pCmdMsg->pMsg, pCxt->pCmdMsg->msgLen, &createSmaReq);
+  tFreeSMCreateSmaReq(&createSmaReq);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -1696,24 +1771,27 @@ static int32_t translateSubquery(STranslateContext* pCxt, SNode* pNode) {
   return code;
 }
 
-static int32_t setReslutSchema(STranslateContext* pCxt, SQuery* pQuery) {
-  if (QUERY_NODE_SELECT_STMT == nodeType(pQuery->pRoot)) {
-    SSelectStmt* pSelect = (SSelectStmt*)pQuery->pRoot;
-    pQuery->numOfResCols = LIST_LENGTH(pSelect->pProjectionList);
-    pQuery->pResSchema = calloc(pQuery->numOfResCols, sizeof(SSchema));
-    if (NULL == pQuery->pResSchema) {
-      return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_OUT_OF_MEMORY);
+int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
+  if (QUERY_NODE_SELECT_STMT == nodeType(pRoot)) {
+    SSelectStmt* pSelect = (SSelectStmt*) pRoot;
+    *numOfCols = LIST_LENGTH(pSelect->pProjectionList);
+    *pSchema = calloc((*numOfCols), sizeof(SSchema));
+    if (NULL == (*pSchema)) {
+      return TSDB_CODE_OUT_OF_MEMORY;
     }
+
     SNode* pNode;
     int32_t index = 0;
     FOREACH(pNode, pSelect->pProjectionList) {
       SExprNode* pExpr = (SExprNode*)pNode;
-      pQuery->pResSchema[index].type = pExpr->resType.type;
-      pQuery->pResSchema[index].bytes = pExpr->resType.bytes;
-      strcpy(pQuery->pResSchema[index].name, pExpr->aliasName);
+      (*pSchema)[index].type = pExpr->resType.type;
+      (*pSchema)[index].bytes = pExpr->resType.bytes;
+      (*pSchema)[index].colId = index + 1;
+      strcpy((*pSchema)[index].name, pExpr->aliasName);
       index +=1;
     }
   }
+
   return TSDB_CODE_SUCCESS;
 }
 
@@ -2297,7 +2375,7 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
       pQuery->haveResultSet = true;
       pQuery->directRpc = false;
       pQuery->msgType = TDMT_VND_QUERY;
-      code = setReslutSchema(pCxt, pQuery);
+      code = qExtractResultSchema(pQuery->pRoot, &pQuery->numOfResCols, &pQuery->pResSchema);
       break;
     case QUERY_NODE_VNODE_MODIF_STMT:
       pQuery->haveResultSet = false;

source/libs/sync/inc/syncRaftEntry.h

@@ -27,6 +27,12 @@ extern "C" {
 #include "syncMessage.h"
 #include "taosdef.h"
 
+typedef enum EntryType {
+  SYNC_RAFT_ENTRY_NOOP = 0,
+  SYNC_RAFT_ENTRY_DATA = 1,
+  SYNC_RAFT_ENTRY_CONFIG = 2,
+} EntryType;
+
 typedef struct SSyncRaftEntry {
   uint32_t  bytes;
   uint32_t  msgType;
@@ -35,12 +41,15 @@ typedef struct SSyncRaftEntry {
   bool      isWeak;
   SyncTerm  term;
   SyncIndex index;
+  EntryType entryType;
   uint32_t  dataLen;
   char      data[];
 } SSyncRaftEntry;
 
 SSyncRaftEntry* syncEntryBuild(uint32_t dataLen);
 SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index);  // step 4
+SSyncRaftEntry* syncEntryBuild3(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index, EntryType entryType);
+SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index);
 void            syncEntryDestory(SSyncRaftEntry* pEntry);
 char*           syncEntrySerialize(const SSyncRaftEntry* pEntry, uint32_t* len);  // step 5
 SSyncRaftEntry* syncEntryDeserialize(const char* buf, uint32_t len);              // step 6

source/libs/sync/src/syncAppendEntries.c

@@ -199,6 +199,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
             SSyncRaftEntry* pRollBackEntry = logStoreGetEntry(ths->pLogStore, index);
             assert(pRollBackEntry != NULL);
 
+            // maybe is a NOOP ENTRY
+            // assert(pRollBackEntry->entryType == SYNC_RAFT_ENTRY_DATA);
+
             SRpcMsg rpcMsg;
             syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg);
             ths->pFsm->FpRollBackCb(ths->pFsm, &rpcMsg, pRollBackEntry->index, pRollBackEntry->isWeak, 0, ths->state);
@@ -217,7 +220,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
         SRpcMsg rpcMsg;
         syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
         if (ths->pFsm != NULL) {
-          if (ths->pFsm->FpPreCommitCb != NULL) {
+          if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
             ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 2, ths->state);
           }
         }
@@ -242,7 +245,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
       SRpcMsg rpcMsg;
       syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
       if (ths->pFsm != NULL) {
-        if (ths->pFsm->FpPreCommitCb != NULL) {
+        if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
           ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pAppendEntry->index, pAppendEntry->isWeak, 3, ths->state);
         }
       }
@@ -298,7 +301,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
               SRpcMsg rpcMsg;
               syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
-              if (ths->pFsm->FpCommitCb != NULL) {
+              if (ths->pFsm->FpCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
                 ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state);
               }
 

source/libs/sync/src/syncCommit.c

@@ -97,7 +97,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
           SRpcMsg rpcMsg;
           syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
-          if (pSyncNode->pFsm->FpCommitCb != NULL) {
+          if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
             pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, pSyncNode->state);
           }
 

source/libs/sync/src/syncIO.c

@@ -29,7 +29,7 @@ static int32_t  syncIODestroy(SSyncIO *io);
 static int32_t  syncIOStartInternal(SSyncIO *io);
 static int32_t  syncIOStopInternal(SSyncIO *io);
 
-static void *  syncIOConsumerFunc(void *param);
+static void   *syncIOConsumerFunc(void *param);
 static void    syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static void    syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet);
 static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
@@ -75,6 +75,7 @@ int32_t syncIOSendMsg(void *clientRpc, const SEpSet *pEpSet, SRpcMsg *pMsg) {
   syncRpcMsgPrint2(logBuf, pMsg);
 
   pMsg->handle = NULL;
+  pMsg->noResp = 1;
   rpcSendRequest(clientRpc, pEpSet, pMsg, NULL);
   return ret;
 }
@@ -234,9 +235,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) {
 }
 
 static void *syncIOConsumerFunc(void *param) {
-  SSyncIO *  io = param;
+  SSyncIO   *io = param;
   STaosQall *qall;
-  SRpcMsg *  pRpcMsg, rpcMsg;
+  SRpcMsg   *pRpcMsg, rpcMsg;
   qall = taosAllocateQall();
 
   while (1) {
@@ -324,19 +325,21 @@ static void *syncIOConsumerFunc(void *param) {
       taosGetQitem(qall, (void **)&pRpcMsg);
       rpcFreeCont(pRpcMsg->pCont);
 
-      if (pRpcMsg->handle != NULL) {
-        int msgSize = 32;
-        memset(&rpcMsg, 0, sizeof(rpcMsg));
-        rpcMsg.msgType = SYNC_RESPONSE;
-        rpcMsg.pCont = rpcMallocCont(msgSize);
-        rpcMsg.contLen = msgSize;
-        snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
-        rpcMsg.handle = pRpcMsg->handle;
-        rpcMsg.code = 0;
-
-        syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
-        rpcSendResponse(&rpcMsg);
-      }
+      /*
+            if (pRpcMsg->handle != NULL) {
+              int msgSize = 32;
+              memset(&rpcMsg, 0, sizeof(rpcMsg));
+              rpcMsg.msgType = SYNC_RESPONSE;
+              rpcMsg.pCont = rpcMallocCont(msgSize);
+              rpcMsg.contLen = msgSize;
+              snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", "give a reply");
+              rpcMsg.handle = pRpcMsg->handle;
+              rpcMsg.code = 0;
+
+              syncRpcMsgPrint2((char *)"syncIOConsumerFunc rpcSendResponse --> ", &rpcMsg);
+              rpcSendResponse(&rpcMsg);
+            }
+      */
 
       taosFreeQitem(pRpcMsg);
     }

source/libs/sync/src/syncMain.c

@@ -37,11 +37,13 @@ static int32_t tsNodeRefId = -1;
 
 // ------ local funciton ---------
 // enqueue message ----
-static void syncNodeEqPingTimer(void* param, void* tmrId);
-static void syncNodeEqElectTimer(void* param, void* tmrId);
-static void syncNodeEqHeartbeatTimer(void* param, void* tmrId);
+static void    syncNodeEqPingTimer(void* param, void* tmrId);
+static void    syncNodeEqElectTimer(void* param, void* tmrId);
+static void    syncNodeEqHeartbeatTimer(void* param, void* tmrId);
+static int32_t syncNodeEqNoop(SSyncNode* ths);
+static int32_t syncNodeAppendNoop(SSyncNode* ths);
 
-// on message ----
+// process message ----
 static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
 static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg);
@@ -669,6 +671,10 @@ void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
   assert(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
   assert(voteGrantedMajority(pSyncNode->pVotesGranted));
   syncNodeBecomeLeader(pSyncNode);
+
+  // Raft 3.6.2 Committing entries from previous terms
+  syncNodeAppendNoop(pSyncNode);
+  // syncNodeEqNoop(pSyncNode);
 }
 
 void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
@@ -803,6 +809,47 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
   }
 }
 
+static int32_t syncNodeEqNoop(SSyncNode* ths) {
+  int32_t ret = 0;
+  assert(ths->state == TAOS_SYNC_STATE_LEADER);
+
+  SyncIndex       index = ths->pLogStore->getLastIndex(ths->pLogStore) + 1;
+  SyncTerm        term = ths->pRaftStore->currentTerm;
+  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index);
+  assert(pEntry != NULL);
+
+  uint32_t           entryLen;
+  char*              serialized = syncEntrySerialize(pEntry, &entryLen);
+  SyncClientRequest* pSyncMsg = syncClientRequestBuild(entryLen);
+  assert(pSyncMsg->dataLen == entryLen);
+  memcpy(pSyncMsg->data, serialized, entryLen);
+
+  SRpcMsg rpcMsg;
+  syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
+  ths->FpEqMsg(ths->queue, &rpcMsg);
+
+  free(serialized);
+  syncClientRequestDestroy(pSyncMsg);
+
+  return ret;
+}
+
+static int32_t syncNodeAppendNoop(SSyncNode* ths) {
+  int32_t ret = 0;
+
+  SyncIndex       index = ths->pLogStore->getLastIndex(ths->pLogStore) + 1;
+  SyncTerm        term = ths->pRaftStore->currentTerm;
+  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index);
+  assert(pEntry != NULL);
+
+  if (ths->state == TAOS_SYNC_STATE_LEADER) {
+    ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
+    syncNodeReplicate(ths);
+  }
+
+  return ret;
+}
+
 // on message ----
 static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
   int32_t ret = 0;
@@ -851,7 +898,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
     syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
     if (ths->pFsm != NULL) {
-      if (ths->pFsm->FpPreCommitCb != NULL) {
+      if (ths->pFsm->FpPreCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
         ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0, ths->state);
       }
     }
@@ -866,7 +913,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
     syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
     if (ths->pFsm != NULL) {
-      if (ths->pFsm->FpPreCommitCb != NULL) {
+      if (ths->pFsm->FpPreCommitCb != NULL && pEntry->entryType == SYNC_RAFT_ENTRY_DATA) {
         ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 1, ths->state);
       }
     }

source/libs/sync/src/syncRaftEntry.c

@@ -28,6 +28,13 @@ SSyncRaftEntry* syncEntryBuild(uint32_t dataLen) {
 
 // step 4. SyncClientRequest => SSyncRaftEntry, add term, index
 SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index) {
+  SSyncRaftEntry* pEntry = syncEntryBuild3(pMsg, term, index, SYNC_RAFT_ENTRY_DATA);
+  assert(pEntry != NULL);
+
+  return pEntry;
+}
+
+SSyncRaftEntry* syncEntryBuild3(SyncClientRequest* pMsg, SyncTerm term, SyncIndex index, EntryType entryType) {
   SSyncRaftEntry* pEntry = syncEntryBuild(pMsg->dataLen);
   assert(pEntry != NULL);
 
@@ -37,12 +44,23 @@ SSyncRaftEntry* syncEntryBuild2(SyncClientRequest* pMsg, SyncTerm term, SyncInde
   pEntry->isWeak = pMsg->isWeak;
   pEntry->term = term;
   pEntry->index = index;
+  pEntry->entryType = entryType;
   pEntry->dataLen = pMsg->dataLen;
   memcpy(pEntry->data, pMsg->data, pMsg->dataLen);
 
   return pEntry;
 }
 
+SSyncRaftEntry* syncEntryBuildNoop(SyncTerm term, SyncIndex index) {
+  SSyncRaftEntry* pEntry = syncEntryBuild(0);
+  assert(pEntry != NULL);
+  pEntry->term = term;
+  pEntry->index = index;
+  pEntry->entryType = SYNC_RAFT_ENTRY_NOOP;
+
+  return pEntry;
+}
+
 void syncEntryDestory(SSyncRaftEntry* pEntry) {
   if (pEntry != NULL) {
     free(pEntry);
@@ -85,6 +103,7 @@ cJSON* syncEntry2Json(const SSyncRaftEntry* pEntry) {
     cJSON_AddStringToObject(pRoot, "term", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%lu", pEntry->index);
     cJSON_AddStringToObject(pRoot, "index", u64buf);
+    cJSON_AddNumberToObject(pRoot, "entryType", pEntry->entryType);
     cJSON_AddNumberToObject(pRoot, "dataLen", pEntry->dataLen);
 
     char* s;

source/libs/sync/src/syncRaftLog.c

@@ -34,7 +34,7 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
   pLogStore->getLastTerm = logStoreLastTerm;
   pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
   pLogStore->getCommitIndex = logStoreGetCommitIndex;
-  return pLogStore;  // to avoid compiler error
+  return pLogStore;
 }
 
 void logStoreDestory(SSyncLogStore* pLogStore) {
@@ -48,18 +48,22 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
 
-  assert(pEntry->index == logStoreLastIndex(pLogStore) + 1);
+  SyncIndex lastIndex = logStoreLastIndex(pLogStore);
+  assert(pEntry->index == lastIndex + 1);
   uint32_t len;
   char*    serialized = syncEntrySerialize(pEntry, &len);
   assert(serialized != NULL);
 
-  int code;
-  code = walWrite(pWal, pEntry->index, pEntry->msgType, serialized, len);
-  assert(code == 0);
+  int code = 0;
+  /*
+    code = walWrite(pWal, pEntry->index, pEntry->entryType, serialized, len);
+    assert(code == 0);
+  */
+  assert(walWrite(pWal, pEntry->index, pEntry->entryType, serialized, len) == 0);
 
   walFsync(pWal, true);
   free(serialized);
-  return code;  // to avoid compiler error
+  return code;
 }
 
 SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
@@ -69,7 +73,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
 
   if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
     SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
-    walReadWithHandle(pWalHandle, index);
+    assert(walReadWithHandle(pWalHandle, index) == 0);
     pEntry = syncEntryDeserialize(pWalHandle->pHead->head.body, pWalHandle->pHead->head.len);
     assert(pEntry != NULL);
 
@@ -83,7 +87,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
 int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  walRollback(pWal, fromIndex);
+  assert(walRollback(pWal, fromIndex) == 0);
   return 0;  // to avoid compiler error
 }
 
@@ -107,7 +111,7 @@ SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
 int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  walCommit(pWal, index);
+  assert(walCommit(pWal, index) == 0);
   return 0;  // to avoid compiler error
 }
 

source/libs/sync/test/CMakeLists.txt

@@ -34,6 +34,7 @@ add_executable(syncEncodeTest "")
 add_executable(syncWriteTest "")
 add_executable(syncReplicateTest "")
 add_executable(syncReplicateTest2 "")
+add_executable(syncReplicateTest3 "")
 add_executable(syncReplicateLoadTest "")
 add_executable(syncRefTest "")
 add_executable(syncLogStoreCheck "")
@@ -183,6 +184,10 @@ target_sources(syncReplicateTest2
     PRIVATE
     "syncReplicateTest2.cpp"
 )
+target_sources(syncReplicateTest3
+    PRIVATE
+    "syncReplicateTest3.cpp"
+)
 target_sources(syncReplicateLoadTest
     PRIVATE
     "syncReplicateLoadTest.cpp"
@@ -377,6 +382,11 @@ target_include_directories(syncReplicateTest2
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncReplicateTest3
+    PUBLIC
+    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 target_include_directories(syncReplicateLoadTest
     PUBLIC
     "${CMAKE_SOURCE_DIR}/include/libs/sync"
@@ -538,6 +548,10 @@ target_link_libraries(syncReplicateTest2
     sync
     gtest_main
 )
+target_link_libraries(syncReplicateTest3
+    sync
+    gtest_main
+)
 target_link_libraries(syncReplicateLoadTest
     sync
     gtest_main

source/libs/sync/test/syncEntryTest.cpp

@@ -46,6 +46,20 @@ void test2() {
 }
 
 void test3() {
+  SyncClientRequest* pSyncMsg = syncClientRequestBuild(10);
+  pSyncMsg->originalRpcType = 33;
+  pSyncMsg->seqNum = 11;
+  pSyncMsg->isWeak = 1;
+  strcpy(pSyncMsg->data, "test3");
+
+  SSyncRaftEntry* pEntry = syncEntryBuild3(pSyncMsg, 100, 200, SYNC_RAFT_ENTRY_NOOP);
+  syncEntryPrint(pEntry);
+
+  syncClientRequestDestroy(pSyncMsg);
+  syncEntryDestory(pEntry);
+}
+
+void test4() {
   SSyncRaftEntry* pEntry = syncEntryBuild(10);
   assert(pEntry != NULL);
   pEntry->msgType = 11;
@@ -54,7 +68,8 @@ void test3() {
   pEntry->isWeak = true;
   pEntry->term = 44;
   pEntry->index = 55;
-  strcpy(pEntry->data, "test3");
+  pEntry->entryType = SYNC_RAFT_ENTRY_CONFIG;
+  strcpy(pEntry->data, "test4");
   syncEntryPrint(pEntry);
 
   uint32_t len;
@@ -76,6 +91,7 @@ int main(int argc, char** argv) {
   test1();
   test2();
   test3();
+  test4();
 
   return 0;
 }

source/libs/sync/test/syncReplicateLoadTest.cpp

@@ -178,9 +178,10 @@ int main(int argc, char **argv) {
   while (1) {
     sTrace(
         "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
-        "electTimerMS:%d",
+        "electTimerMS:%d, commitIndex:%ld",
         pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
-        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS);
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
+        pSyncNode->commitIndex);
     taosMsleep(1000);
   }
 

source/libs/sync/test/syncReplicateTest.cpp

@@ -183,18 +183,20 @@ int main(int argc, char **argv) {
 
     taosMsleep(1000);
     sTrace(
-        "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
-        "electTimerMS:%d",
+        "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
+        "electTimerMS:%d, commitIndex:%ld",
         gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
-        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS,
+        gSyncNode->commitIndex);
   }
 
   while (1) {
     sTrace(
         "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
-        "electTimerMS:%d",
+        "electTimerMS:%d, commitIndex:%ld",
         gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm,
-        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS);
+        gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS,
+        gSyncNode->commitIndex);
     taosMsleep(1000);
   }
 

source/libs/sync/test/syncReplicateTest2.cpp

@@ -185,17 +185,19 @@ int main(int argc, char **argv) {
 
     sTrace(
         "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
-        "electTimerMS:%d",
+        "electTimerMS:%d, commitIndex:%ld",
         pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
-        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS);
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
+        pSyncNode->commitIndex);
   }
 
   while (1) {
     sTrace(
         "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
-        "electTimerMS:%d",
+        "electTimerMS:%d, commitIndex:%ld",
         pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
-        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS);
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
+        pSyncNode->commitIndex);
     taosMsleep(1000);
   }
 

source/libs/sync/test/syncReplicateTest3.cpp

+#define ALLOW_FORBID_FUNC
+
+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncEnv.h"
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncMessage.h"
+#include "syncRaftEntry.h"
+#include "syncRaftLog.h"
+#include "syncRaftStore.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+uint16_t ports[] = {7010, 7110, 7210, 7310, 7410};
+int32_t  replicaNum = 3;
+int32_t  myIndex = 0;
+
+SRaftId   ids[TSDB_MAX_REPLICA];
+SSyncInfo syncInfo;
+SSyncFSM *pFsm;
+SWal *    pWal;
+
+void CommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
+              ESyncState state) {
+  char logBuf[256];
+  snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n",
+           pFsm, index, isWeak, code, state, syncUtilState2String(state));
+  syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
+}
+
+void PreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
+                 ESyncState state) {
+  char logBuf[256];
+  snprintf(logBuf, sizeof(logBuf),
+           "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, index, isWeak,
+           code, state, syncUtilState2String(state));
+  syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
+}
+
+void RollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code,
+                ESyncState state) {
+  char logBuf[256];
+  snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n",
+           pFsm, index, isWeak, code, state, syncUtilState2String(state));
+  syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg);
+}
+
+void initFsm() {
+  pFsm = (SSyncFSM *)malloc(sizeof(SSyncFSM));
+  pFsm->FpCommitCb = CommitCb;
+  pFsm->FpPreCommitCb = PreCommitCb;
+  pFsm->FpRollBackCb = RollBackCb;
+}
+
+int64_t syncNodeInit() {
+  syncInfo.vgId = 1234;
+  syncInfo.rpcClient = gSyncIO->clientRpc;
+  syncInfo.FpSendMsg = syncIOSendMsg;
+  syncInfo.queue = gSyncIO->pMsgQ;
+  syncInfo.FpEqMsg = syncIOEqMsg;
+  syncInfo.pFsm = pFsm;
+  snprintf(syncInfo.path, sizeof(syncInfo.path), "./replicate2_test_%d", myIndex);
+
+  int code = walInit();
+  assert(code == 0);
+  SWalCfg walCfg;
+  memset(&walCfg, 0, sizeof(SWalCfg));
+  walCfg.vgId = syncInfo.vgId;
+  walCfg.fsyncPeriod = 1000;
+  walCfg.retentionPeriod = 1000;
+  walCfg.rollPeriod = 1000;
+  walCfg.retentionSize = 1000;
+  walCfg.segSize = 1000;
+  walCfg.level = TAOS_WAL_FSYNC;
+
+  char tmpdir[128];
+  snprintf(tmpdir, sizeof(tmpdir), "./replicate2_test_wal_%d", myIndex);
+  pWal = walOpen(tmpdir, &walCfg);
+  assert(pWal != NULL);
+
+  syncInfo.pWal = pWal;
+
+  SSyncCfg *pCfg = &syncInfo.syncCfg;
+  pCfg->myIndex = myIndex;
+  pCfg->replicaNum = replicaNum;
+
+  for (int i = 0; i < replicaNum; ++i) {
+    pCfg->nodeInfo[i].nodePort = ports[i];
+    snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
+    // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
+  }
+
+  int64_t rid = syncStart(&syncInfo);
+  assert(rid > 0);
+
+  SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid);
+  assert(pSyncNode != NULL);
+
+  // pSyncNode->hbBaseLine = 500;
+  // pSyncNode->electBaseLine = 1500;
+
+  gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing;
+  gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply;
+  gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote;
+  gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply;
+  gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries;
+  gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply;
+  gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout;
+  gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest;
+  gSyncIO->pSyncNode = pSyncNode;
+
+  syncNodeRelease(pSyncNode);
+
+  return rid;
+}
+
+void initRaftId(SSyncNode *pSyncNode) {
+  for (int i = 0; i < replicaNum; ++i) {
+    ids[i] = pSyncNode->replicasId[i];
+    char *s = syncUtilRaftId2Str(&ids[i]);
+    printf("raftId[%d] : %s\n", i, s);
+    free(s);
+  }
+}
+
+SRpcMsg *step0(int i) {
+  SRpcMsg *pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg));
+  memset(pMsg, 0, sizeof(SRpcMsg));
+  pMsg->msgType = 9999;
+  pMsg->contLen = 128;
+  pMsg->pCont = malloc(pMsg->contLen);
+  snprintf((char *)(pMsg->pCont), pMsg->contLen, "value-%u-%d", ports[myIndex], i);
+  return pMsg;
+}
+
+SyncClientRequest *step1(const SRpcMsg *pMsg) {
+  SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true);
+  return pRetMsg;
+}
+
+int main(int argc, char **argv) {
+  // taosInitLog((char *)"syncTest.log", 100000, 10);
+  tsAsyncLog = 0;
+  sDebugFlag = 143 + 64;
+  void logTest();
+
+  myIndex = 0;
+  if (argc >= 2) {
+    myIndex = atoi(argv[1]);
+  }
+
+  int recordCount = 100;
+  if (argc >= 3) {
+    recordCount = atoi(argv[2]);
+  }
+
+  int sleepMS = 10;
+  if (argc >= 4) {
+    sleepMS = atoi(argv[3]);
+  }
+
+  int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]);
+  assert(ret == 0);
+
+  initFsm();
+
+  ret = syncInit();
+  assert(ret == 0);
+
+  int64_t rid = syncNodeInit();
+  assert(rid > 0);
+
+  SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid);
+  assert(pSyncNode != NULL);
+
+  syncNodePrint2((char *)"", pSyncNode);
+  initRaftId(pSyncNode);
+
+  for (int i = 0; i < recordCount; ++i) {
+    // step0
+    SRpcMsg *pMsg0 = step0(i);
+    syncRpcMsgPrint2((char *)"==step0==", pMsg0);
+
+    syncPropose(rid, pMsg0, true);
+    taosMsleep(sleepMS);
+
+    free(pMsg0);
+
+    sTrace(
+        "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
+        "electTimerMS:%d, commitIndex:%ld",
+        pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
+        pSyncNode->commitIndex);
+  }
+
+  while (1) {
+    sTrace(
+        "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
+        "electTimerMS:%d, commitIndex:%ld",
+        pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
+        pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS,
+        pSyncNode->commitIndex);
+    taosMsleep(1000);
+  }
+
+  return 0;
+}

tests/script/jenkins/basic.txt

@@ -21,7 +21,7 @@
 
 # ---- query
 ./test.sh -f tsim/query/interval.sim
-#./test.sh -f tsim/query/interval-offset.sim
+./test.sh -f tsim/query/interval-offset.sim
 
 # ---- show
 ./test.sh -f tsim/show/basic.sim 

tests/script/tsim/insert/basic0.sim

@@ -109,7 +109,26 @@ endi
 
 #print =============== select first(*)/first(column) from child table
 #sql select first(*) from ct1
-#sql select first(ts), first(c1), first(c2), first(c3) from ct1
+#print ====> select first(*) from ct1
+#print rows: $rows 
+#print $data00  $data01 $data02 $data03
+
+sql select first(ts), first(c1), first(c2), first(c3) from ct1
+print ====> select first(ts), first(c1), first(c2), first(c3) from ct1
+print rows: $rows 
+print $data00  $data01 $data02 $data03
+if $rows != 1 then 
+  return -1
+endi
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
 
 print =============== select min(column) from child table
 sql select min(c1), min(c2), min(c3) from ct1
@@ -216,10 +235,42 @@ endi
 #  return -1
 #endi
 
+print =============== select count(column) from supter table
+sql select ts, c1, c2, c3 from stb
+print rows: $rows
+print $data00 $data01 $data02  $data03
+print $data10 $data11 $data12  $data13
+print $data20 $data21 $data22  $data23
+print $data30 $data31 $data32  $data33
+print $data40 $data41 $data42  $data43
+print $data50 $data51 $data52  $data53
+print $data60 $data61 $data62  $data63
+print $data70 $data71 $data72  $data73
+print $data80 $data81 $data82  $data83
+if $rows != 9 then 
+  return -1
+endi
+
+# The order of data from different sub tables in the super table is random, 
+# so this detection may fail randomly
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
+
 #print =============== select count(column) from supter table
 #sql select count(ts), count(c1), count(c2), count(c3) from stb
+#print rows: $rows
 #print $data00 $data01 $data02  $data03
-#if $data00 != 8 then 
+#print $data10 $data11 $data12  $data13
+#print $data20 $data21 $data22  $data23
+#print $data30 $data31 $data32  $data33
+#if $data00 != 9 then 
 #  return -1
 #endi
 #if $data01 != 8 then 
@@ -232,7 +283,6 @@ endi
 #  return -1
 #endi
 
-
 #===================================================================
 #===================================================================
 
@@ -317,7 +367,26 @@ endi
 
 #print =============== select first(*)/first(column) from child table
 #sql select first(*) from ct1
-#sql select first(ts), first(c1), first(c2), first(c3) from ct1
+#print ====> select first(*) from ct1
+#print rows: $rows 
+#print $data00  $data01 $data02 $data03
+
+sql select first(ts), first(c1), first(c2), first(c3) from ct1
+print ====> select first(ts), first(c1), first(c2), first(c3) from ct1
+print rows: $rows 
+print $data00  $data01 $data02 $data03
+if $rows != 1 then 
+  return -1
+endi
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
 
 print =============== select min(column) from child table
 sql select min(c1), min(c2), min(c3) from ct1
@@ -424,6 +493,33 @@ endi
 #  return -1
 #endi
 
+print =============== select count(column) from supter table
+sql select ts, c1, c2, c3 from stb
+print rows: $rows
+print $data00 $data01 $data02  $data03
+print $data10 $data11 $data12  $data13
+print $data20 $data21 $data22  $data23
+print $data30 $data31 $data32  $data33
+print $data40 $data41 $data42  $data43
+print $data50 $data51 $data52  $data53
+print $data60 $data61 $data62  $data63
+print $data70 $data71 $data72  $data73
+print $data80 $data81 $data82  $data83
+if $rows != 9 then 
+  return -1
+endi
+# The order of data from different sub tables in the super table is random, 
+# so this detection may fail randomly
+if $data01 != 10 then 
+  return -1
+endi
+if $data02 != 2.00000 then 
+  return -1
+endi
+if $data03 != 3.000000000 then 
+  return -1
+endi
+
 #print =============== select count(column) from supter table
 #sql select count(ts), count(c1), count(c2), count(c3) from stb
 #print $data00 $data01 $data02  $data03

tests/script/tsim/query/interval-offset.sim

@@ -88,14 +88,17 @@ print ===> rows4: $data40 $data41 $data42 $data43 $data44
 print ===> rows5: $data50 $data51 $data52 $data53 $data54
 print ===> rows6: $data60 $data61 $data62 $data63 $data64
 print ===> rows7: $data70 $data71 $data72 $data73 $data74
-if $rows != 8 then
-  print expect 8, actual $rows
+print ===> rows8: $data80 $data81 $data82 $data83 $data84
+if $rows != 9 then
   return -1
 endi 
-if $data00 != 2 then
+if $data00 != 1 then
+  return -1
+endi 
+if $data70 != 2 then
   return -1
 endi 
-if $data70 != 1 then
+if $data80 != 1 then
   return -1
 endi 
 
@@ -166,15 +169,15 @@ print ===> rows7: $data70 $data71 $data72 $data73 $data74 $data75
 #endi
 
 print =============== insert data into child table ct3 (n)
-sql insert into ct3 values ( '2021-12-21 01:01:01.000', NULL )
-sql insert into ct3 values ( '2021-12-31 01:01:01.000', 1 )
-sql insert into ct3 values ( '2022-01-01 01:01:06.000', 2 )
-sql insert into ct3 values ( '2022-01-07 01:01:10.000', 3 )
-sql insert into ct3 values ( '2022-01-31 01:01:16.000', 4 )
-sql insert into ct3 values ( '2022-02-01 01:01:20.000', 5 )
-sql insert into ct3 values ( '2022-02-28 01:01:26.000', 6 )
-sql insert into ct3 values ( '2022-03-01 01:01:30.000', 7 )
-sql insert into ct3 values ( '2022-03-08 01:01:36.000', 8 )
+sql insert into ct3 values ( '2021-12-21 01:01:01.000', NULL );
+sql insert into ct3 values ( '2021-12-31 01:01:01.000', 1 );
+sql insert into ct3 values ( '2022-01-01 01:01:06.000', 2 );
+sql insert into ct3 values ( '2022-01-07 01:01:10.000', 3 );
+sql insert into ct3 values ( '2022-01-31 01:01:16.000', 4 );
+sql insert into ct3 values ( '2022-02-01 01:01:20.000', 5 );
+sql insert into ct3 values ( '2022-02-28 01:01:26.000', 6 );
+sql insert into ct3 values ( '2022-03-01 01:01:30.000', 7 );
+sql insert into ct3 values ( '2022-03-08 01:01:36.000', 8 );
 
 sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct3 interval(1n, 1w)
 print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(*) from ct3 interval(1n, 1w)

tools/shell/src/shellMain.c

@@ -17,6 +17,7 @@
 #include "os.h"
 #include "shell.h"
 #include "tglobal.h"
+#include "tconfig.h"
 #include "shellCommand.h"
 #include "tbase64.h"
 #include "tlog.h"
@@ -619,20 +620,17 @@ int main(int argc, char *argv[]) {
 
   shellParseArgument(argc, argv, &args);
 
-#if 0
   if (args.dump_config) {
-    taosInitGlobalCfg();
-    taosReadGlobalLogCfg();
+    taosInitCfg(configDir, NULL, NULL, NULL, 1);
 
-    if (taosReadGlobalCfg() ! =0) {
-      printf("TDengine read global config failed");
+    SConfig *pCfg = taosGetCfg();
+    if (NULL == pCfg) {
+      printf("TDengine read global config failed!\n");
       exit(EXIT_FAILURE);
     }
-
-    taosDumpGlobalCfg();
+    cfgDumpCfg(pCfg, 0, 1);
     exit(0);
   }
-#endif
 
   if (args.netTestRole && args.netTestRole[0] != 0) {
     TAOS *con = NULL;