Merge branch '3.0' into enh/stopquery

include/common/ttokendef.h

@@ -216,51 +216,55 @@
 #define TK_NOW            198
 #define TK_TODAY          199
 #define TK_TIMEZONE       200
-#define TK_COUNT         201
-#define TK_LAST_ROW      202
-#define TK_BETWEEN       203
-#define TK_IS            204
-#define TK_NK_LT         205
-#define TK_NK_GT         206
-#define TK_NK_LE         207
-#define TK_NK_GE         208
-#define TK_NK_NE         209
-#define TK_MATCH         210
-#define TK_NMATCH        211
-#define TK_CONTAINS      212
-#define TK_JOIN          213
-#define TK_INNER         214
-#define TK_SELECT        215
-#define TK_DISTINCT      216
-#define TK_WHERE         217
-#define TK_PARTITION     218
-#define TK_BY            219
-#define TK_SESSION       220
-#define TK_STATE_WINDOW  221
-#define TK_SLIDING       222
-#define TK_FILL          223
-#define TK_VALUE         224
-#define TK_NONE          225
-#define TK_PREV          226
-#define TK_LINEAR        227
-#define TK_NEXT          228
-#define TK_HAVING        229
-#define TK_RANGE         230
-#define TK_EVERY         231
-#define TK_ORDER         232
-#define TK_SLIMIT        233
-#define TK_SOFFSET       234
-#define TK_LIMIT         235
-#define TK_OFFSET        236
-#define TK_ASC           237
-#define TK_NULLS         238
-#define TK_ID            239
-#define TK_NK_BITNOT     240
-#define TK_INSERT        241
-#define TK_VALUES        242
-#define TK_IMPORT        243
-#define TK_NK_SEMI       244
-#define TK_FILE          245
+#define TK_CLIENT_VERSION 201
+#define TK_SERVER_VERSION 202
+#define TK_SERVER_STATUS  203
+#define TK_CURRENT_USER   204
+#define TK_COUNT          205
+#define TK_LAST_ROW       206
+#define TK_BETWEEN        207
+#define TK_IS             208
+#define TK_NK_LT          209
+#define TK_NK_GT          210
+#define TK_NK_LE          211
+#define TK_NK_GE          212
+#define TK_NK_NE          213
+#define TK_MATCH          214
+#define TK_NMATCH         215
+#define TK_CONTAINS       216
+#define TK_JOIN           217
+#define TK_INNER          218
+#define TK_SELECT         219
+#define TK_DISTINCT       220
+#define TK_WHERE          221
+#define TK_PARTITION      222
+#define TK_BY             223
+#define TK_SESSION        224
+#define TK_STATE_WINDOW   225
+#define TK_SLIDING        226
+#define TK_FILL           227
+#define TK_VALUE          228
+#define TK_NONE           229
+#define TK_PREV           230
+#define TK_LINEAR         231
+#define TK_NEXT           232
+#define TK_HAVING         233
+#define TK_RANGE          234
+#define TK_EVERY          235
+#define TK_ORDER          236
+#define TK_SLIMIT         237
+#define TK_SOFFSET        238
+#define TK_LIMIT          239
+#define TK_OFFSET         240
+#define TK_ASC            241
+#define TK_NULLS          242
+#define TK_ID             243
+#define TK_NK_BITNOT      244
+#define TK_INSERT         245
+#define TK_VALUES         246
+#define TK_IMPORT         247
+#define TK_NK_SEMI        248
+#define TK_FILE           249
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/executor/executor.h

@@ -36,7 +36,6 @@ typedef struct SReadHandle {
   void*   vnode;
   void*   mnd;
   SMsgCb* pMsgCb;
-//  int8_t  initTsdbReader;
 } SReadHandle;
 
 enum {
@@ -140,12 +139,6 @@ int32_t qKillTask(qTaskInfo_t tinfo);
  */
 int32_t qAsyncKillTask(qTaskInfo_t tinfo);
 
-/**
- * return whether query is completed or not
- * @param tinfo
- * @return
- */
-int32_t qIsTaskCompleted(qTaskInfo_t tinfo);
 
 /**
  * destroy query info structure
@@ -176,6 +169,15 @@ int32_t qSerializeTaskStatus(qTaskInfo_t tinfo, char** pOutput, int32_t* len);
 
 int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t len);
 
+/**
+ * return the scan info, in the form of tuple of two items, including table uid and current timestamp
+ * @param tinfo
+ * @param uid
+ * @param ts
+ * @return
+ */
+int32_t qGetStreamScanStatus(qTaskInfo_t tinfo, uint64_t* uid, int64_t* ts);
+
 #ifdef __cplusplus
 }
 #endif

include/libs/function/functionMgt.h

@@ -105,7 +105,7 @@ typedef enum EFunctionType {
   // system function
   FUNCTION_TYPE_DATABASE = 3000,
   FUNCTION_TYPE_CLIENT_VERSION,
-  FUNCTION_TYPE_SERVER_SERSION,
+  FUNCTION_TYPE_SERVER_VERSION,
   FUNCTION_TYPE_SERVER_STATUS,
   FUNCTION_TYPE_CURRENT_USER,
   FUNCTION_TYPE_USER,
@@ -193,6 +193,7 @@ bool fmIsForbidGroupByFunc(int32_t funcId);
 bool fmIsIntervalInterpoFunc(int32_t funcId);
 bool fmIsInterpFunc(int32_t funcId);
 bool fmIsLastRowFunc(int32_t funcId);
+bool fmIsSystemInfoFunc(int32_t funcId);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
 

include/libs/nodes/querynodes.h

@@ -89,6 +89,7 @@ typedef struct SValueNode {
   bool      isDuration;
   bool      translate;
   bool      notReserved;
+  bool      isNull;
   int16_t   placeholderNo;
   union {
     bool     b;

include/libs/planner/planner.h

@@ -48,7 +48,7 @@ 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);
+void qClearSubplanExecutionNode(SSubplan* pSubplan);
 
 // Convert to subplan to string for the scheduler to send to the executor
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);

include/libs/transport/trpc.h

@@ -45,7 +45,7 @@ typedef struct SRpcHandleInfo {
   int32_t  noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int32_t  persistHandle;  // persist handle or not
   STraceId traceId;
-  // int64_t traceId;
+  int8_t   hasEpSet;
 
   // app info
   void *ahandle;  // app handle set by client
@@ -123,7 +123,7 @@ void *  rpcReallocCont(void *ptr, int32_t contLen);
 void rpcSendRequest(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid);
 void rpcSendResponse(const SRpcMsg *pMsg);
 void rpcRegisterBrokenLinkArg(SRpcMsg *msg);
-void rpcReleaseHandle(void *handle, int8_t type);  // just release client conn to rpc instance, no close sock
+void rpcReleaseHandle(void *handle, int8_t type);  // just release conn to rpc instance, no close sock
 
 // These functions will not be called in the child process
 void    rpcSendRedirectRsp(void *pConn, const SEpSet *pEpSet);

source/client/src/clientMain.c

@@ -740,6 +740,8 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
       .schemalessType = pTscObj->schemalessType,
       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
       .async = true,
+      .svrVer = pTscObj->sVer,
+      .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes)
   };
   return TSDB_CODE_SUCCESS;
 }

source/common/src/tdatablock.c

@@ -1662,8 +1662,8 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
   int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
   int32_t rows = pDataBlock->info.rows;
   int32_t len = 0;
-  len += snprintf(dumpBuf + len, size - len, "\n%s |block type %d |child id %d|\n", flag,
-                  (int32_t)pDataBlock->info.type, pDataBlock->info.childId);
+  len += snprintf(dumpBuf + len, size - len, "\n%s |block type %d |child id %d|group id %lu|\n", flag,
+                  (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.groupId);
   for (int32_t j = 0; j < rows; j++) {
     len += snprintf(dumpBuf + len, size - len, "%s |", flag);
     for (int32_t k = 0; k < colNum; k++) {

source/common/src/tmsg.c

@@ -621,8 +621,8 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
     if (NULL == pReq->pFuncs) return -1;
   }
   for (int32_t i = 0; i < numOfFuncs; ++i) {
-    char *pFunc = NULL;
-    if (tDecodeCStrAlloc(&decoder, &pFunc) < 0) return -1;
+    char pFunc[TSDB_FUNC_NAME_LEN] = {0};
+    if (tDecodeCStrTo(&decoder, pFunc) < 0) return -1;
     if (taosArrayPush(pReq->pFuncs, pFunc) == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
@@ -2300,7 +2300,6 @@ int32_t tDeserializeSServerVerRsp(void *buf, int32_t bufLen, SServerVerRsp *pRsp
   return 0;
 }
 
-
 int32_t tSerializeSQnodeListRsp(void *buf, int32_t bufLen, SQnodeListRsp *pRsp) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -2387,7 +2386,6 @@ int32_t tDeserializeSDnodeListRsp(void *buf, int32_t bufLen, SDnodeListRsp *pRsp
 
 void tFreeSDnodeListRsp(SDnodeListRsp *pRsp) { taosArrayDestroy(pRsp->dnodeList); }
 
-
 int32_t tSerializeSCompactDbReq(void *buf, int32_t bufLen, SCompactDbReq *pReq) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -2909,20 +2907,19 @@ int32_t tDeserializeSShowVariablesReq(void *buf, int32_t bufLen, SShowVariablesR
   return 0;
 }
 
-int32_t tEncodeSVariablesInfo(SEncoder* pEncoder, SVariablesInfo* pInfo) {
+int32_t tEncodeSVariablesInfo(SEncoder *pEncoder, SVariablesInfo *pInfo) {
   if (tEncodeCStr(pEncoder, pInfo->name) < 0) return -1;
   if (tEncodeCStr(pEncoder, pInfo->value) < 0) return -1;
   return 0;
 }
 
-int32_t tDecodeSVariablesInfo(SDecoder* pDecoder, SVariablesInfo* pInfo) {
+int32_t tDecodeSVariablesInfo(SDecoder *pDecoder, SVariablesInfo *pInfo) {
   if (tDecodeCStrTo(pDecoder, pInfo->name) < 0) return -1;
   if (tDecodeCStrTo(pDecoder, pInfo->value) < 0) return -1;
   return 0;
 }
 
-
-int32_t tSerializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pRsp) {
+int32_t tSerializeSShowVariablesRsp(void *buf, int32_t bufLen, SShowVariablesRsp *pRsp) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
 
@@ -2930,7 +2927,7 @@ int32_t tSerializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp
   int32_t varNum = taosArrayGetSize(pRsp->variables);
   if (tEncodeI32(&encoder, varNum) < 0) return -1;
   for (int32_t i = 0; i < varNum; ++i) {
-    SVariablesInfo* pInfo = taosArrayGet(pRsp->variables, i);
+    SVariablesInfo *pInfo = taosArrayGet(pRsp->variables, i);
     if (tEncodeSVariablesInfo(&encoder, pInfo) < 0) return -1;
   }
   tEndEncode(&encoder);
@@ -2940,7 +2937,7 @@ int32_t tSerializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp
   return tlen;
 }
 
-int32_t tDeserializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesRsp* pRsp) {
+int32_t tDeserializeSShowVariablesRsp(void *buf, int32_t bufLen, SShowVariablesRsp *pRsp) {
   SDecoder decoder = {0};
   tDecoderInit(&decoder, buf, bufLen);
 
@@ -2962,7 +2959,7 @@ int32_t tDeserializeSShowVariablesRsp(void* buf, int32_t bufLen, SShowVariablesR
   return 0;
 }
 
-void tFreeSShowVariablesRsp(SShowVariablesRsp* pRsp) {
+void tFreeSShowVariablesRsp(SShowVariablesRsp *pRsp) {
   if (NULL == pRsp) {
     return;
   }

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

@@ -136,7 +136,7 @@ int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
       if (vnodeProcessDropTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
       break;
     case TDMT_VND_DROP_TTL_TABLE:
-      if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
+      //if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
       break;
     case TDMT_VND_CREATE_SMA: {
       if (vnodeProcessCreateTSmaReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;

source/libs/command/src/command.c

@@ -18,7 +18,31 @@
 #include "tdatablock.h"
 #include "tglobal.h"
 
-extern SConfig *tsCfg;
+extern SConfig* tsCfg;
+
+static int32_t buildRetrieveTableRsp(SSDataBlock* pBlock, int32_t numOfCols, SRetrieveTableRsp** pRsp) {
+  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
+  *pRsp = taosMemoryCalloc(1, rspSize);
+  if (NULL == *pRsp) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  (*pRsp)->useconds = 0;
+  (*pRsp)->completed = 1;
+  (*pRsp)->precision = 0;
+  (*pRsp)->compressed = 0;
+  (*pRsp)->compLen = 0;
+  (*pRsp)->numOfRows = htonl(pBlock->info.rows);
+  (*pRsp)->numOfCols = htonl(numOfCols);
+
+  int32_t len = 0;
+  blockCompressEncode(pBlock, (*pRsp)->data, &len, numOfCols, false);
+  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
+
+  blockDataDestroy(pBlock);
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t getSchemaBytes(const SSchema* pSchema) {
   switch (pSchema->type) {
     case TSDB_DATA_TYPE_BINARY:
@@ -89,31 +113,11 @@ static int32_t execDescribe(SNode* pStmt, SRetrieveTableRsp** pRsp) {
   SSDataBlock* pBlock = buildDescResultDataBlock();
   setDescResultIntoDataBlock(pBlock, numOfRows, pDesc->pMeta);
 
-  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
-  *pRsp = taosMemoryCalloc(1, rspSize);
-  if (NULL == *pRsp) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  (*pRsp)->useconds = 0;
-  (*pRsp)->completed = 1;
-  (*pRsp)->precision = 0;
-  (*pRsp)->compressed = 0;
-  (*pRsp)->compLen = 0;
-  (*pRsp)->numOfRows = htonl(numOfRows);
-  (*pRsp)->numOfCols = htonl(DESCRIBE_RESULT_COLS);
-
-  int32_t len = 0;
-  blockCompressEncode(pBlock, (*pRsp)->data, &len, DESCRIBE_RESULT_COLS, false);
-  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
-
-  blockDataDestroy(pBlock);
-  return TSDB_CODE_SUCCESS;
+  return buildRetrieveTableRsp(pBlock, DESCRIBE_RESULT_COLS, pRsp);
 }
 
 static int32_t execResetQueryCache() { return catalogClearCache(); }
 
-
 static SSDataBlock* buildCreateDBResultDataBlock() {
   SSDataBlock*    pBlock = createDataBlock();
   SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, SHOW_CREATE_DB_RESULT_COLS, 1);
@@ -149,14 +153,14 @@ int64_t getValOfDiffPrecision(int8_t unit, int64_t val) {
   return v;
 }
 
-char *buildRetension(SArray *pRetension) {
+char* buildRetension(SArray* pRetension) {
   size_t size = taosArrayGetSize(pRetension);
   if (size == 0) {
     return NULL;
   }
 
-  char       *p1 = taosMemoryCalloc(1, 100);
-  SRetention *p = taosArrayGet(pRetension, 0);
+  char*       p1 = taosMemoryCalloc(1, 100);
+  SRetention* p = taosArrayGet(pRetension, 0);
 
   int32_t len = 0;
 
@@ -185,8 +189,7 @@ char *buildRetension(SArray *pRetension) {
   return p1;
 }
 
-
-static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char *dbFName, SDbCfgInfo* pCfg) {
+static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbFName, SDbCfgInfo* pCfg) {
   blockDataEnsureCapacity(pBlock, 1);
   pBlock->info.rows = 1;
 
@@ -198,7 +201,7 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char *dbFName, S
   SColumnInfoData* pCol2 = taosArrayGet(pBlock->pDataBlock, 1);
   char             buf2[SHOW_CREATE_DB_RESULT_FIELD2_LEN] = {0};
   int32_t          len = 0;
-  char *prec = NULL;
+  char*            prec = NULL;
   switch (pCfg->precision) {
     case TSDB_TIME_PRECISION_MILLI:
       prec = TSDB_TIME_PRECISION_MILLI_STR;
@@ -214,14 +217,15 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char *dbFName, S
       break;
   }
 
-  char *retentions = buildRetension(pCfg->pRetensions);
+  char* retentions = buildRetension(pCfg->pRetensions);
 
-  len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE DATABASE `%s` BUFFER %d CACHELAST %d COMP %d DURATION %dm "
+  len += sprintf(buf2 + VARSTR_HEADER_SIZE,
+                 "CREATE DATABASE `%s` BUFFER %d CACHELAST %d COMP %d DURATION %dm "
                  "FSYNC %d MAXROWS %d MINROWS %d KEEP %dm,%dm,%dm PAGES %d PAGESIZE %d PRECISION '%s' REPLICA %d "
                  "STRICT %d WAL %d VGROUPS %d SINGLE_STABLE %d",
-                    dbFName, pCfg->buffer, pCfg->cacheLastRow, pCfg->compression, pCfg->daysPerFile,
-                    pCfg->fsyncPeriod, pCfg->maxRows, pCfg->minRows, pCfg->daysToKeep0, pCfg->daysToKeep1, pCfg->daysToKeep2,
-                    pCfg->pages, pCfg->pageSize, prec, pCfg->replications, pCfg->strict, pCfg->walLevel, pCfg->numOfVgroups,
+                 dbFName, pCfg->buffer, pCfg->cacheLastRow, pCfg->compression, pCfg->daysPerFile, pCfg->fsyncPeriod,
+                 pCfg->maxRows, pCfg->minRows, pCfg->daysToKeep0, pCfg->daysToKeep1, pCfg->daysToKeep2, pCfg->pages,
+                 pCfg->pageSize, prec, pCfg->replications, pCfg->strict, pCfg->walLevel, pCfg->numOfVgroups,
                  1 == pCfg->numOfStables);
 
   if (retentions) {
@@ -234,31 +238,10 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char *dbFName, S
   colDataAppend(pCol2, 0, buf2, false);
 }
 
-
 static int32_t execShowCreateDatabase(SShowCreateDatabaseStmt* pStmt, SRetrieveTableRsp** pRsp) {
   SSDataBlock* pBlock = buildCreateDBResultDataBlock();
   setCreateDBResultIntoDataBlock(pBlock, pStmt->dbName, pStmt->pCfg);
-
-  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
-  *pRsp = taosMemoryCalloc(1, rspSize);
-  if (NULL == *pRsp) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  (*pRsp)->useconds = 0;
-  (*pRsp)->completed = 1;
-  (*pRsp)->precision = 0;
-  (*pRsp)->compressed = 0;
-  (*pRsp)->compLen = 0;
-  (*pRsp)->numOfRows = htonl(1);
-  (*pRsp)->numOfCols = htonl(SHOW_CREATE_DB_RESULT_COLS);
-
-  int32_t len = 0;
-  blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_CREATE_DB_RESULT_COLS, false);
-  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
-
-  blockDataDestroy(pBlock);
-  return TSDB_CODE_SUCCESS;
+  return buildRetrieveTableRsp(pBlock, SHOW_CREATE_DB_RESULT_COLS, pRsp);
 }
 
 static SSDataBlock* buildCreateTbResultDataBlock() {
@@ -281,7 +264,7 @@ void appendColumnFields(char* buf, int32_t* len, STableCfg* pCfg) {
     if (TSDB_DATA_TYPE_VARCHAR == pSchema->type) {
       sprintf(type + strlen(type), "(%d)", (int32_t)(pSchema->bytes - VARSTR_HEADER_SIZE));
     } else if (TSDB_DATA_TYPE_NCHAR == pSchema->type) {
-      sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE));    
+      sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
     }
 
     *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s`%s` %s", ((i > 0) ? ", " : ""), pSchema->name, type);
@@ -296,14 +279,13 @@ void appendTagFields(char* buf, int32_t* len, STableCfg* pCfg) {
     if (TSDB_DATA_TYPE_VARCHAR == pSchema->type) {
       sprintf(type + strlen(type), "(%d)", (int32_t)(pSchema->bytes - VARSTR_HEADER_SIZE));
     } else if (TSDB_DATA_TYPE_NCHAR == pSchema->type) {
-      sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE));    
+      sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
     }
 
     *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s`%s` %s", ((i > 0) ? ", " : ""), pSchema->name, type);
   }
 }
 
-
 void appendTagNameFields(char* buf, int32_t* len, STableCfg* pCfg) {
   for (int32_t i = 0; i < pCfg->numOfTags; ++i) {
     SSchema* pSchema = pCfg->pSchemas + pCfg->numOfColumns + i;
@@ -311,13 +293,12 @@ void appendTagNameFields(char* buf, int32_t* len, STableCfg* pCfg) {
   }
 }
 
-
 int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
-  SArray *pTagVals = NULL;
-  STag *pTag = (STag*)pCfg->pTags;
+  SArray* pTagVals = NULL;
+  STag*   pTag = (STag*)pCfg->pTags;
 
   if (pCfg->pTags && pTag->flags & TD_TAG_JSON) {
-    char *pJson = parseTagDatatoJson(pTag);
+    char* pJson = parseTagDatatoJson(pTag);
     if (pJson) {
       *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s", pJson);
       taosMemoryFree(pJson);
@@ -326,7 +307,7 @@ int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t code = tTagToValArray((const STag *)pCfg->pTags, &pTagVals);
+  int32_t code = tTagToValArray((const STag*)pCfg->pTags, &pTagVals);
   if (code) {
     return code;
   }
@@ -345,7 +326,7 @@ int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
       continue;
     }
 
-    STagVal *pTagVal = (STagVal *)taosArrayGet(pTagVals, j);
+    STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);
     if (pSchema->colId > pTagVal->cid) {
       qError("tag value and column mismatch, schemaId:%d, valId:%d", pSchema->colId, pTagVal->cid);
       taosArrayDestroy(pTagVals);
@@ -365,7 +346,6 @@ int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
       *len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "NULL");
     }
 
-
     /*
     if (type == TSDB_DATA_TYPE_BINARY) {
       if (pTagVal->nData > 0) {
@@ -436,7 +416,7 @@ void appendTableOptions(char* buf, int32_t* len, STableCfg* pCfg) {
   }
 }
 
-static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, char *tbName, STableCfg* pCfg) {
+static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, char* tbName, STableCfg* pCfg) {
   int32_t code = 0;
   blockDataEnsureCapacity(pBlock, 1);
   pBlock->info.rows = 1;
@@ -480,34 +460,13 @@ static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, char *tbName,
   return TSDB_CODE_SUCCESS;
 }
 
-
 static int32_t execShowCreateTable(SShowCreateTableStmt* pStmt, SRetrieveTableRsp** pRsp) {
   SSDataBlock* pBlock = buildCreateTbResultDataBlock();
   int32_t      code = setCreateTBResultIntoDataBlock(pBlock, pStmt->tableName, pStmt->pCfg);
   if (code) {
     return code;
   }
-
-  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
-  *pRsp = taosMemoryCalloc(1, rspSize);
-  if (NULL == *pRsp) {
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  (*pRsp)->useconds = 0;
-  (*pRsp)->completed = 1;
-  (*pRsp)->precision = 0;
-  (*pRsp)->compressed = 0;
-  (*pRsp)->compLen = 0;
-  (*pRsp)->numOfRows = htonl(1);
-  (*pRsp)->numOfCols = htonl(SHOW_CREATE_TB_RESULT_COLS);
-
-  int32_t len = 0;
-  blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_CREATE_TB_RESULT_COLS, false);
-  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
-
-  blockDataDestroy(pBlock);
-  return TSDB_CODE_SUCCESS;
+  return buildRetrieveTableRsp(pBlock, SHOW_CREATE_TB_RESULT_COLS, pRsp);
 }
 
 static int32_t execShowCreateSTable(SShowCreateTableStmt* pStmt, SRetrieveTableRsp** pRsp) {
@@ -551,18 +510,17 @@ static SSDataBlock* buildLocalVariablesResultDataBlock() {
   return pBlock;
 }
 
-
 int32_t setLocalVariablesResultIntoDataBlock(SSDataBlock* pBlock) {
   int32_t numOfCfg = taosArrayGetSize(tsCfg->array);
   int32_t numOfRows = 0;
   blockDataEnsureCapacity(pBlock, numOfCfg);
 
   for (int32_t i = 0, c = 0; i < numOfCfg; ++i, c = 0) {
-    SConfigItem *pItem = taosArrayGet(tsCfg->array, i);
+    SConfigItem* pItem = taosArrayGet(tsCfg->array, i);
 
     char name[TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE] = {0};
     STR_WITH_MAXSIZE_TO_VARSTR(name, pItem->name, TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE);
-    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
+    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
     colDataAppend(pColInfo, i, name, false);
 
     char    value[TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};
@@ -575,42 +533,70 @@ int32_t setLocalVariablesResultIntoDataBlock(SSDataBlock* pBlock) {
     numOfRows++;
   }
 
-
   pBlock->info.rows = numOfRows;
 
   return TSDB_CODE_SUCCESS;
 }
 
-
 static int32_t execShowLocalVariables(SRetrieveTableRsp** pRsp) {
   SSDataBlock* pBlock = buildLocalVariablesResultDataBlock();
   int32_t      code = setLocalVariablesResultIntoDataBlock(pBlock);
   if (code) {
     return code;
   }
+  return buildRetrieveTableRsp(pBlock, SHOW_LOCAL_VARIABLES_RESULT_COLS, pRsp);
+}
 
-  size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
-  *pRsp = taosMemoryCalloc(1, rspSize);
-  if (NULL == *pRsp) {
+static int32_t createSelectResultDataBlock(SNodeList* pProjects, SSDataBlock** pOutput) {
+  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
+  if (NULL == pBlock) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  (*pRsp)->useconds = 0;
-  (*pRsp)->completed = 1;
-  (*pRsp)->precision = 0;
-  (*pRsp)->compressed = 0;
-  (*pRsp)->compLen = 0;
-  (*pRsp)->numOfRows = htonl(pBlock->info.rows);
-  (*pRsp)->numOfCols = htonl(SHOW_LOCAL_VARIABLES_RESULT_COLS);
+  pBlock->pDataBlock = taosArrayInit(LIST_LENGTH(pProjects), sizeof(SColumnInfoData));
 
-  int32_t len = 0;
-  blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_LOCAL_VARIABLES_RESULT_COLS, false);
-  ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
+  SNode* pProj = NULL;
+  FOREACH(pProj, pProjects) {
+    SColumnInfoData infoData = {0};
+    infoData.info.type = ((SExprNode*)pProj)->resType.type;
+    infoData.info.bytes = ((SExprNode*)pProj)->resType.bytes;
+    taosArrayPush(pBlock->pDataBlock, &infoData);
+  }
+  *pOutput = pBlock;
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t buildSelectResultDataBlock(SNodeList* pProjects, SSDataBlock* pBlock) {
+  int32_t numOfCols = LIST_LENGTH(pProjects);
+  blockDataEnsureCapacity(pBlock, 1);
+
+  int32_t index = 0;
+  SNode*  pProj = NULL;
+  FOREACH(pProj, pProjects) {
+    if (((SValueNode*)pProj)->isNull) {
+      colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, NULL, true);
+    } else {
+      colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, nodesGetValueFromNode((SValueNode*)pProj), false);
+    }
+  }
+
+  pBlock->info.rows = 1;
 
-  blockDataDestroy(pBlock);
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t execSelectWithoutFrom(SSelectStmt* pSelect, SRetrieveTableRsp** pRsp) {
+  SSDataBlock* pBlock = NULL;
+  int32_t      code = createSelectResultDataBlock(pSelect->pProjectionList, &pBlock);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = buildSelectResultDataBlock(pSelect->pProjectionList, pBlock);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = buildRetrieveTableRsp(pBlock, LIST_LENGTH(pSelect->pProjectionList), pRsp);
+  }
+  return code;
+}
+
 int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp) {
   switch (nodeType(pStmt)) {
     case QUERY_NODE_DESCRIBE_STMT:
@@ -627,6 +613,8 @@ int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp) {
       return execAlterLocal((SAlterLocalStmt*)pStmt);
     case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
       return execShowLocalVariables(pRsp);
+    case QUERY_NODE_SELECT_STMT:
+      return execSelectWithoutFrom((SSelectStmt*)pStmt, pRsp);
     default:
       break;
   }

source/libs/executor/inc/executorimpl.h

@@ -253,18 +253,15 @@ typedef struct STableScanInfo {
   SReadHandle     readHandle;
 
   SFileBlockLoadRecorder readRecorder;
-  int64_t         numOfRows;
   SScanInfo       scanInfo;
   int32_t         scanTimes;
   SNode*          pFilterNode;  // filter info, which is push down by optimizer
-  SqlFunctionCtx* pCtx;         // which belongs to the direct upstream operator operator query context
-  SResultRowInfo* pResultRowInfo;
-  int32_t*        rowEntryInfoOffset;
-  SExprInfo*      pExpr;
+  SqlFunctionCtx* pCtx;         // which belongs to the direct upstream operator operator query context,todo: remove this by using SExprSup
+  int32_t*        rowEntryInfoOffset;  // todo: remove this by using SExprSup
+  SExprInfo*      pExpr;// todo: remove this by using SExprSup
+
   SSDataBlock*    pResBlock;
   SArray*         pColMatchInfo;
-  int32_t         numOfOutput;
-
   SExprSupp       pseudoSup;
   SQueryTableDataCond cond;
   int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
@@ -275,8 +272,13 @@ typedef struct STableScanInfo {
   int32_t         curTWinIdx;
 
   int32_t         currentGroupId;
-  uint64_t        queryId;
-  uint64_t        taskId;
+  uint64_t        queryId;   // todo remove it
+  uint64_t        taskId;    // todo remove it
+
+  struct {
+    uint64_t uid;
+    int64_t t;
+  } scanStatus;
 } STableScanInfo;
 
 typedef struct STagScanInfo {
@@ -321,31 +323,31 @@ typedef struct SessionWindowSupporter {
 } SessionWindowSupporter;
 
 typedef struct SStreamBlockScanInfo {
+  uint64_t        tableUid;         // queried super table uid
+  SExprInfo*      pPseudoExpr;
+  int32_t         numOfPseudoExpr;
+  int32_t         primaryTsIndex;    // primary time stamp slot id
+  SReadHandle     readHandle;
+  SInterval       interval;     // if the upstream is an interval operator, the interval info is also kept here.
+  SArray*         pColMatchInfo;    //
+  SNode*          pCondition;
+
   SArray*         pBlockLists;      // multiple SSDatablock.
   SSDataBlock*    pRes;             // result SSDataBlock
   SSDataBlock*    pUpdateRes;       // update SSDataBlock
   int32_t         updateResIndex;
   int32_t         blockType;        // current block type
   int32_t         validBlockIndex;  // Is current data has returned?
-  SColumnInfo*    pCols;            // the output column info
   uint64_t        numOfExec;        // execution times
   void*           streamBlockReader;// stream block reader handle
-  SArray*         pColMatchInfo;    //
-  SNode*          pCondition;
+
   int32_t         tsArrayIndex;
   SArray*         tsArray;
   uint64_t        groupId;
   SUpdateInfo*    pUpdateInfo;
 
-  SExprInfo*      pPseudoExpr;
-  int32_t         numOfPseudoExpr;
-
-  int32_t         primaryTsIndex;    // primary time stamp slot id
-  SReadHandle     readHandle;
-  uint64_t        tableUid;         // queried super table uid
   EStreamScanMode scanMode;
   SOperatorInfo* pSnapshotReadOp;
-  SInterval      interval;     // if the upstream is an interval operator, the interval info is also kept here.
   SArray*        childIds;
   SessionWindowSupporter sessionSup;
   bool            assignBlockUid; // assign block uid to groupId, temporarily used for generating rollup SMA.
@@ -416,6 +418,7 @@ typedef struct SIntervalAggOperatorInfo {
   STimeWindowAggSupp twAggSup;
   bool               invertible;
   SArray*            pPrevValues;        //  SArray<SGroupKeys> used to keep the previous not null value for interpolation.
+  bool               ignoreCloseWindow;
 } SIntervalAggOperatorInfo;
 
 typedef struct SStreamFinalIntervalOperatorInfo {
@@ -437,6 +440,7 @@ typedef struct SStreamFinalIntervalOperatorInfo {
   SArray*            pPullWins;          // SPullWindowInfo
   int32_t            pullIndex;
   SSDataBlock*       pPullDataRes;
+  bool               ignoreCloseWindow;
 } SStreamFinalIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -574,6 +578,7 @@ typedef struct SStreamSessionAggOperatorInfo {
   SArray*              pChildren;       // cache for children's result; final stream operator
   SPhysiNode*          pPhyNode;           // create new child
   bool                 isFinal;
+  bool                 ignoreCloseWindow;
 } SStreamSessionAggOperatorInfo;
 
 typedef struct STimeSliceOperatorInfo {
@@ -617,6 +622,7 @@ typedef struct SStreamStateAggOperatorInfo {
   void*                pDelIterator;
   SArray*              pScanWindow;
   SArray*              pChildren;       // cache for children's result;
+  bool                 ignoreCloseWindow;
 } SStreamStateAggOperatorInfo;
 
 typedef struct SSortedMergeOperatorInfo {
@@ -683,7 +689,7 @@ int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t
 void    initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock);
 void    cleanupBasicInfo(SOptrBasicInfo* pInfo);
 int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr);
-void    cleanupExprSup(SExprSupp* pSup);
+void    cleanupExprSupp(SExprSupp* pSup);
 int32_t initAggInfo(SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                     const char* pkey);
 void    initResultSizeInfo(SOperatorInfo* pOperator, int32_t numOfRows);
@@ -707,7 +713,7 @@ void    destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
 void    appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
 void    setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId);
 
-void    cleanupExecSupp(SExprSupp* pSupp);
+int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts);
 
 SSDataBlock* loadNextDataBlock(void* param);
 

source/libs/executor/src/executorMain.c

@@ -191,16 +191,6 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
-  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
-
-  if (pTaskInfo == NULL) {
-    return TSDB_CODE_QRY_INVALID_QHANDLE;
-  }
-
-  return isTaskKilled(pTaskInfo);
-}
-
 void qDestroyTask(qTaskInfo_t qTaskHandle) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle;
   qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);
@@ -242,3 +232,10 @@ int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t le
 }
 
 
+int32_t qGetStreamScanStatus(qTaskInfo_t tinfo, uint64_t* uid, int64_t* ts) {
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+

source/libs/executor/src/executorimpl.c

@@ -1033,7 +1033,7 @@ static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSData
   SqlFunctionCtx* pCtx = pTableScanInfo->pCtx;
   uint32_t        status = BLK_DATA_NOT_LOAD;
 
-  int32_t numOfOutput = pTableScanInfo->numOfOutput;
+  int32_t numOfOutput = 0;//pTableScanInfo->numOfOutput;
   for (int32_t i = 0; i < numOfOutput; ++i) {
     int32_t functionId = pCtx[i].functionId;
     int32_t colId = pTableScanInfo->pExpr[i].base.pParam[0].pCol->colId;
@@ -2821,6 +2821,24 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
   }
 }
 
+int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts) {
+  int32_t type = pOperator->operatorType;
+  if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+    SStreamBlockScanInfo* pScanInfo = pOperator->info;
+    STableScanInfo* pSnapShotScanInfo = pScanInfo->pSnapshotReadOp->info;
+    *uid = pSnapShotScanInfo->scanStatus.uid;
+    *ts = pSnapShotScanInfo->scanStatus.t;
+  } else {
+    if (pOperator->pDownstream[0] == NULL) {
+      return TSDB_CODE_INVALID_PARA;
+    } else {
+      doGetScanStatus(pOperator->pDownstream[0], uid, ts);
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 // this is a blocking operator
 static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
   if (OPTR_IS_OPENED(pOperator)) {
@@ -3543,7 +3561,7 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
   taosArrayDestroy(pInfo->pPseudoColInfo);
 }
 
-void cleanupExecSupp(SExprSupp* pSupp) {
+void cleanupExprSupp(SExprSupp* pSupp) {
   destroySqlFunctionCtx(pSupp->pCtx, pSupp->numOfExprs);
   destroyExprInfo(pSupp->pExprInfo, pSupp->numOfExprs);
 
@@ -3556,7 +3574,7 @@ static void destroyIndefinitOperatorInfo(void* param, int32_t numOfOutput) {
 
   taosArrayDestroy(pInfo->pPseudoColInfo);
   cleanupAggSup(&pInfo->aggSup);
-  cleanupExecSupp(&pInfo->scalarSup);
+  cleanupExprSupp(&pInfo->scalarSup);
 }
 
 void destroyExchangeOperatorInfo(void* param, int32_t numOfOutput) {

source/libs/executor/src/groupoperator.c

@@ -37,7 +37,7 @@ static void destroyGroupOperatorInfo(void* param, int32_t numOfOutput) {
   taosMemoryFreeClear(pInfo->keyBuf);
   taosArrayDestroy(pInfo->pGroupCols);
   taosArrayDestroy(pInfo->pGroupColVals);
-  cleanupExecSupp(&pInfo->scalarSup);
+  cleanupExprSupp(&pInfo->scalarSup);
 }
 
 static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) {
@@ -701,7 +701,7 @@ static void destroyPartitionOperatorInfo(void* param, int32_t numOfOutput) {
   taosHashCleanup(pInfo->pGroupSet);
   taosMemoryFree(pInfo->columnOffset);
 
-  cleanupExecSupp(&pInfo->scalarSup);
+  cleanupExprSupp(&pInfo->scalarSup);
 }
 
 SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo) {

source/libs/executor/src/scanoperator.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <executorimpl.h>
 #include <vnode.h>
 #include "filter.h"
 #include "function.h"
@@ -413,6 +414,11 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
     pTableScanInfo->readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;
 
     pOperator->cost.totalCost = pTableScanInfo->readRecorder.elapsedTime;
+
+    // todo refactor
+    pTableScanInfo->scanStatus.uid = pBlock->info.uid;
+    pTableScanInfo->scanStatus.t = pBlock->info.window.ekey;
+
     return pBlock;
   }
   return NULL;
@@ -459,7 +465,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
   int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
   if (pTableScanInfo->scanTimes < total) {
     if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
-      prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, pTableScanInfo->numOfOutput);
+      prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, 0);
       tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
       pTableScanInfo->curTWinIdx = 0;
     }

source/libs/executor/src/timewindowoperator.c

@@ -813,6 +813,16 @@ static void removeResults(SArray* pWins, SArray* pUpdated) {
   }
 }
 
+
+bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup) {
+  ASSERT(pSup->maxTs == INT64_MIN || pSup->maxTs > 0);
+  return pSup->maxTs != INT64_MIN && ts < pSup->maxTs - pSup->waterMark;
+}
+
+bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) {
+  return isOverdue(pWin->ekey, pSup);
+}
+
 static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock,
                             int32_t scanFlag, SArray* pUpdated) {
   SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
@@ -830,15 +840,16 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
 
   STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
                                         pInfo->interval.precision, &pInfo->win);
-
-  int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
+  int32_t ret = TSDB_CODE_SUCCESS;
+  if (!pInfo->ignoreCloseWindow || !isCloseWindow(&win, &pInfo->twAggSup)) {
+    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                         pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
     if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
       longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
     }
 
-  if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM) {
-    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
+    if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && 
+        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
       saveResultRow(pResult, tableGroupId, pUpdated);
     }
   }
@@ -864,9 +875,11 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
     doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup);
   }
 
+  if (!pInfo->ignoreCloseWindow || !isCloseWindow(&win, &pInfo->twAggSup)) {
     updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true);
     doApplyFunctions(pTaskInfo, pSup->pCtx, &win, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                      pBlock->info.rows, numOfOutput, pInfo->order);
+  }
 
   doCloseWindow(pResultRowInfo, pInfo, pResult);
 
@@ -877,6 +890,12 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
     if (startPos < 0) {
       break;
     }
+    if (pInfo->ignoreCloseWindow && isCloseWindow(&nextWin, &pInfo->twAggSup)) {
+      ekey = ascScan ? nextWin.ekey : nextWin.skey;
+      forwardRows =
+          getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
+      continue;
+    }
 
     // null data, failed to allocate more memory buffer
     int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
@@ -885,11 +904,10 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
       longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
     }
 
-    if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM) {
-      if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
+    if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && 
+        pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
       saveResultRow(pResult, tableGroupId, pUpdated);
     }
-    }
 
     ekey = ascScan ? nextWin.ekey : nextWin.skey;
     forwardRows =
@@ -1292,11 +1310,6 @@ static int32_t getAllIntervalWindow(SHashObj* pHashMap, SArray* resWins) {
   return TSDB_CODE_SUCCESS;
 }
 
-bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) {
-  ASSERT(pSup->maxTs == INT64_MIN || pSup->maxTs > 0);
-  return pSup->maxTs != INT64_MIN && pWin->ekey < pSup->maxTs - pSup->waterMark;
-}
-
 static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
     SInterval* pInterval, SHashObj* pPullDataMap, SArray* closeWins) {
   void*  pIte = NULL;
@@ -1411,7 +1424,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
   doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
 
   pOperator->status = OP_RES_TO_RETURN;
-
+  printDataBlock(pInfo->binfo.pRes, "single interval");
   return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
 }
 
@@ -1521,6 +1534,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
   pInfo->interval  = *pInterval;
   pInfo->execModel = pTaskInfo->execModel;
   pInfo->twAggSup  = *pTwAggSupp;
+  pInfo->ignoreCloseWindow = false;
 
   if (pPhyNode->window.pExprs != NULL) {
     int32_t    numOfScalar = 0;
@@ -2276,7 +2290,15 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
   STimeWindow nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
                                             pInfo->interval.precision, NULL);
   while (1) {
-    if (IS_FINAL_OP(pInfo) && isCloseWindow(&nextWin, &pInfo->twAggSup) && pInfo->pChildren) {
+    bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
+    if (pInfo->ignoreCloseWindow && isClosed) {
+      startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
+      if (startPos < 0) {
+        break;
+      }
+      continue;
+    }
+    if (IS_FINAL_OP(pInfo) && isClosed && pInfo->pChildren) {
       bool ignore = true;
       SWinRes winRes = {.ts = nextWin.skey, .groupId = tableGroupId,};
       void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
@@ -2684,6 +2706,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
   _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
   pInfo->pPullDataMap = taosHashInit(64, hashFn, false, HASH_NO_LOCK);
   pInfo->pPullDataRes = createPullDataBlock();
+  pInfo->ignoreCloseWindow = false;
 
   pOperator->operatorType = pPhyNode->type;
   pOperator->blocking = true;
@@ -2830,6 +2853,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
   pInfo->pChildren = NULL;
   pInfo->isFinal = false;
   pInfo->pPhyNode = pPhyNode;
+  pInfo->ignoreCloseWindow = false;
 
   pOperator->name = "StreamSessionWindowAggOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
@@ -3007,6 +3031,9 @@ static int32_t doOneWindowAggImpl(int32_t tsColId, SOptrBasicInfo* pBinfo, SStre
   updateTimeWindowInfo(pTimeWindowData, &pCurWin->win, false);
   doApplyFunctions(pTaskInfo, pSup->pCtx, &pCurWin->win, pTimeWindowData, startIndex, winRows, tsCols,
                    pSDataBlock->info.rows, numOutput, TSDB_ORDER_ASC);
+  SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, pCurWin->pos.pageId);
+  setBufPageDirty(bufPage, true);
+  releaseBufPage(pAggSup->pResultBuf, bufPage);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -3063,7 +3090,13 @@ void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex,
       pWinInfo->isOutput = false;
     }
     taosArrayRemove(pInfo->streamAggSup.pCurWins, i);
+    SFilePage* tmpPage = getBufPage(pInfo->streamAggSup.pResultBuf, pWinInfo->pos.pageId);
+    releaseBufPage(pInfo->streamAggSup.pResultBuf, tmpPage);
   }
+  SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pCurWin->pos.pageId);
+  ASSERT(num > 0);
+  setBufPageDirty(bufPage, true);
+  releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
 }
 
 static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pStUpdated,
@@ -3083,7 +3116,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
   SResultRow* pResult = NULL;
   int32_t     winRows = 0;
 
-  if (pSDataBlock->pDataBlock != NULL) {
+  ASSERT(pSDataBlock->pDataBlock);
   SColumnInfoData* pStartTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
   startTsCols = (int64_t*)pStartTsCol->pData;
   SColumnInfoData* pEndTsCol = NULL;
@@ -3093,12 +3126,13 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
     pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
   }
   endTsCols = (int64_t*)pEndTsCol->pData;
-  } else {
-    return;
-  }
 
   SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
   for (int32_t i = 0; i < pSDataBlock->info.rows;) {
+    if (pInfo->ignoreCloseWindow && isOverdue(endTsCols[i], &pInfo->twAggSup)) {
+      i++;
+      continue;
+    }
     int32_t            winIndex = 0;
     SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, startTsCols[i], endTsCols[i], groupId, gap, &winIndex);
     winRows =
@@ -3205,17 +3239,24 @@ static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWin
         index = 0;
       }
       for (int32_t k = index; k < chWinSize; k++) {
-        SResultWindowInfo* pcw = taosArrayGet(pChWins, k);
-        if (pParentWin->win.skey <= pcw->win.skey && pcw->win.ekey <= pParentWin->win.ekey) {
+        SResultWindowInfo* pChWin = taosArrayGet(pChWins, k);
+        if (pParentWin->win.skey <= pChWin->win.skey && pChWin->win.ekey <= pParentWin->win.ekey) {
           SResultRow* pChResult = NULL;
-          setWindowOutputBuf(pcw, &pChResult, pChild->exprSupp.pCtx, groupId, numOfOutput,
+          setWindowOutputBuf(pChWin, &pChResult, pChild->exprSupp.pCtx, groupId, numOfOutput,
                              pChild->exprSupp.rowEntryInfoOffset, &pChInfo->streamAggSup, pTaskInfo);
           compactFunctions(pSup->pCtx, pChild->exprSupp.pCtx, numOfOutput, pTaskInfo);
+          SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pChWin->pos.pageId);
+          setBufPageDirty(bufPage, true);
+          releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
           continue;
         }
         break;
       }
     }
+    SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pParentWin->pos.pageId);
+    ASSERT(size > 0);
+    setBufPageDirty(bufPage, true);
+    releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
   }
 }
 
@@ -3234,7 +3275,7 @@ int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArra
     for (int32_t i = 0; i < size; i++) {
       void*              pWin = taosArrayGet(pWins, i);
       SResultWindowInfo* pSeWin = fn(pWin);
-      if (pSeWin->win.ekey < pTwSup->maxTs - pTwSup->waterMark) {
+      if (isCloseWindow(&pSeWin->win, pTwSup)) {
         if (!pSeWin->isClosed) {
           pSeWin->isClosed = true;
           if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
@@ -3745,6 +3786,10 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
   SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
   SColumnInfoData*     pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId);
   for (int32_t i = 0; i < pSDataBlock->info.rows; i += winRows) {
+    if (pInfo->ignoreCloseWindow && isOverdue(tsCols[i], &pInfo->twAggSup)) {
+      i++;
+      continue;
+    }
     char*             pKeyData = colDataGetData(pKeyColInfo, i);
     int32_t           winIndex = 0;
     bool              allEqual = true;
@@ -3895,6 +3940,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
   pInfo->pDelRes->info.type = STREAM_DELETE;
   blockDataEnsureCapacity(pInfo->pDelRes, 64);
   pInfo->pChildren = NULL;
+  pInfo->ignoreCloseWindow = false;
 
   pOperator->name = "StreamStateAggOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;

source/libs/function/inc/functionMgtInt.h

@@ -46,6 +46,7 @@ extern "C" {
 #define FUNC_MGT_FORBID_STREAM_FUNC     FUNC_MGT_FUNC_CLASSIFICATION_MASK(17)
 #define FUNC_MGT_FORBID_WINDOW_FUNC     FUNC_MGT_FUNC_CLASSIFICATION_MASK(18)
 #define FUNC_MGT_FORBID_GROUP_BY_FUNC   FUNC_MGT_FUNC_CLASSIFICATION_MASK(19)
+#define FUNC_MGT_SYSTEM_INFO_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(20)
 
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 

source/libs/function/src/builtins.c

@@ -1539,6 +1539,36 @@ static int32_t translateGroupKey(SFunctionNode* pFunc, char* pErrBuf, int32_t le
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t translateDatabaseFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = TSDB_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateClientVersionFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = TSDB_VERSION_LEN, .type = TSDB_DATA_TYPE_VARCHAR};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateServerVersionFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = TSDB_VERSION_LEN, .type = TSDB_DATA_TYPE_VARCHAR};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateServerStatusFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes, .type = TSDB_DATA_TYPE_INT};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateCurrentUserFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = TSDB_USER_LEN, .type = TSDB_DATA_TYPE_VARCHAR};
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t translateUserFunc(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
+  pFunc->node.resType = (SDataType){.bytes = TSDB_USER_LEN, .type = TSDB_DATA_TYPE_VARCHAR};
+  return TSDB_CODE_SUCCESS;
+}
+
 // clang-format off
 const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
@@ -1904,7 +1934,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_first_partial",
     .type = FUNCTION_TYPE_FIRST_PARTIAL,
-    .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 = translateFirstLastPartial,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -1915,7 +1945,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_first_merge",
     .type = FUNCTION_TYPE_FIRST_MERGE,
-    .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 = translateFirstLastMerge,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -1939,7 +1969,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_last_partial",
     .type = FUNCTION_TYPE_LAST_PARTIAL,
-    .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 = translateFirstLastPartial,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -1950,7 +1980,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
   {
     .name = "_last_merge",
     .type = FUNCTION_TYPE_LAST_MERGE,
-    .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 = translateFirstLastMerge,
     .getEnvFunc   = getFirstLastFuncEnv,
     .initFunc     = functionSetup,
@@ -2546,6 +2576,42 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .pPartialFunc = "_group_key",
     .pMergeFunc   = "_group_key"
   },
+  {
+    .name = "database",
+    .type = FUNCTION_TYPE_DATABASE,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateDatabaseFunc,
+  },
+  {
+    .name = "client_version",
+    .type = FUNCTION_TYPE_CLIENT_VERSION,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateClientVersionFunc,
+  },
+  {
+    .name = "server_version",
+    .type = FUNCTION_TYPE_SERVER_VERSION,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateServerVersionFunc,
+  },
+  {
+    .name = "server_status",
+    .type = FUNCTION_TYPE_SERVER_STATUS,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateServerStatusFunc,
+  },
+  {
+    .name = "current_user",
+    .type = FUNCTION_TYPE_CURRENT_USER,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateCurrentUserFunc,
+  },
+  {
+    .name = "user",
+    .type = FUNCTION_TYPE_USER,
+    .classification = FUNC_MGT_SYSTEM_INFO_FUNC | FUNC_MGT_SCALAR_FUNC,
+    .translateFunc = translateUserFunc,
+  },
 };
 // clang-format on
 

source/libs/function/src/builtinsimpl.c

@@ -2409,11 +2409,11 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t getFirstLastInfoSize(int32_t resBytes) { return sizeof(SFirstLastRes) + resBytes + sizeof(int64_t); }
+int32_t getFirstLastInfoSize(int32_t resBytes) { return sizeof(SFirstLastRes) + resBytes + sizeof(int64_t) + sizeof(STuplePos); }
 
 bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   SColumnNode* pNode = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);
-  pEnv->calcMemSize = sizeof(SFirstLastRes) + pNode->node.resType.bytes + sizeof(int64_t);
+  pEnv->calcMemSize = getFirstLastInfoSize(pNode->node.resType.bytes);
   return true;
 }
 
@@ -2491,9 +2491,17 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
         }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
         pInfo->hasResult = true;
-        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
-
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
         pResInfo->numOfRes = 1;
         break;
       }
@@ -2525,8 +2533,17 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
         }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
         pInfo->hasResult = true;
-        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
         pResInfo->numOfRes = 1;
         break;
       }
@@ -2580,8 +2597,17 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
         }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
-        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
         pInfo->hasResult = true;
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
         pResInfo->numOfRes = 1;
       }
       break;
@@ -2603,9 +2629,18 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
         }
         memcpy(pInfo->buf, data, bytes);
         *(TSKEY*)(pInfo->buf + bytes) = cts;
+        //handle selectivity
+        if (pCtx->subsidiaries.num > 0) {
+          STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
+          if (!pInfo->hasResult) {
+            saveTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          } else {
+            copyTupleData(pCtx, i, pCtx->pSrcBlock, pTuplePos);
+          }
+        }
         pInfo->hasResult = true;
         pResInfo->numOfRes = 1;
-        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
+        //DO_UPDATE_TAG_COLUMNS(pCtx, ts);
       }
       break;
     }
@@ -2615,7 +2650,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void firstLastTransferInfo(SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) {
+static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) {
+  SInputColumnInfoData* pColInfo = &pCtx->input;
+  int32_t start = pColInfo->startRowIndex;
+
   pOutput->bytes = pInput->bytes;
   TSKEY* tsIn = (TSKEY*)(pInput->buf + pInput->bytes);
   TSKEY* tsOut = (TSKEY*)(pOutput->buf + pInput->bytes);
@@ -2632,7 +2670,17 @@ static void firstLastTransferInfo(SFirstLastRes* pInput, SFirstLastRes* pOutput,
   }
   *tsOut = *tsIn;
   memcpy(pOutput->buf, pInput->buf, pOutput->bytes);
+  //handle selectivity
+  STuplePos* pTuplePos = (STuplePos*)(pOutput->buf + pOutput->bytes + sizeof(TSKEY));
+  if (pCtx->subsidiaries.num > 0) {
+    if (!pOutput->hasResult) {
+      saveTupleData(pCtx, start, pCtx->pSrcBlock, pTuplePos);
+    } else {
+      copyTupleData(pCtx, start, pCtx->pSrcBlock, pTuplePos);
+    }
+  }
   pOutput->hasResult = true;
+
   return;
 }
 
@@ -2647,7 +2695,7 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer
   char*          data = colDataGetData(pCol, start);
   SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data);
 
-  firstLastTransferInfo(pInputInfo, pInfo, isFirstQuery);
+  firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery);
 
   int32_t numOfElems = pInputInfo->hasResult ? 1 : 0;
 
@@ -2669,6 +2717,9 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
 
   SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo);
   colDataAppend(pCol, pBlock->info.rows, pRes->buf, pResInfo->isNullRes);
+  //handle selectivity
+  STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
+  setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
 
   return pResInfo->numOfRes;
 }
@@ -2687,6 +2738,9 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   colDataAppend(pCol, pBlock->info.rows, res, false);
+  //handle selectivity
+  STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
+  setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
 
   taosMemoryFree(res);
   return 1;
@@ -3043,7 +3097,9 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
     pItem->uid = uid;
 
     // save the data of this tuple
+    if (pCtx->subsidiaries.num > 0) {
       saveTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
+    }
 
     // allocate the buffer and keep the data of this row into the new allocated buffer
     pEntryInfo->numOfRes++;
@@ -3062,7 +3118,10 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
       pItem->uid = uid;
 
       // save the data of this tuple by over writing the old data
+      if (pCtx->subsidiaries.num > 0) {
         copyTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
+      }
+
       taosheapadjust((void*)pItems, sizeof(STopBotResItem), 0, pEntryInfo->numOfRes - 1, (const void*)&type,
                      topBotResComparFn, NULL, !isTopQuery);
     }

source/libs/function/src/functionMgt.c

@@ -179,6 +179,8 @@ bool fmIsForbidWindowFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId
 
 bool fmIsForbidGroupByFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_GROUP_BY_FUNC); }
 
+bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }
+
 bool fmIsInterpFunc(int32_t funcId) {
   if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
     return false;

source/libs/parser/inc/sql.y

@@ -685,6 +685,12 @@ literal_func(A) ::= NOW(B).
 noarg_func(A) ::= NOW(B).                                                         { A = B; }
 noarg_func(A) ::= TODAY(B).                                                       { A = B; }
 noarg_func(A) ::= TIMEZONE(B).                                                    { A = B; }
+noarg_func(A) ::= DATABASE(B).                                                    { A = B; }
+noarg_func(A) ::= CLIENT_VERSION(B).                                              { A = B; }
+noarg_func(A) ::= SERVER_VERSION(B).                                              { A = B; }
+noarg_func(A) ::= SERVER_STATUS(B).                                               { A = B; }
+noarg_func(A) ::= CURRENT_USER(B).                                                { A = B; }
+noarg_func(A) ::= USER(B).                                                        { A = B; }
 
 %type star_func                                                                   { SToken }
 %destructor star_func                                                             { }

source/libs/parser/src/parInsert.c

@@ -67,6 +67,13 @@ typedef struct SInsertParseContext {
   SParseMetaCache*   pMetaCache;
 } SInsertParseContext;
 
+typedef struct SInsertParseSyntaxCxt {
+  SParseContext*   pComCxt;
+  char*            pSql;
+  SMsgBuf          msg;
+  SParseMetaCache* pMetaCache;
+} SInsertParseSyntaxCxt;
+
 typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
 
 static uint8_t TRUE_VALUE = (uint8_t)TSDB_TRUE;
@@ -1098,11 +1105,24 @@ static int32_t storeTableMeta(SInsertParseContext* pCxt, SHashObj* pHash, SName*
   return taosHashPut(pHash, pName, len, &pBackup, POINTER_BYTES);
 }
 
+static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt);
+
+// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
+static int32_t ignoreAutoCreateTableClause(SInsertParseContext* pCxt) {
+  SToken sToken;
+  NEXT_TOKEN(pCxt->pSql, sToken);
+  SInsertParseSyntaxCxt cxt = {.pComCxt = pCxt->pComCxt, .pSql = pCxt->pSql, .msg = pCxt->msg, .pMetaCache = NULL};
+  int32_t               code = skipUsingClause(&cxt);
+  pCxt->pSql = cxt.pSql;
+  return code;
+}
+
 // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
 static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tbFName) {
   int32_t      len = strlen(tbFName);
   STableMeta** pMeta = taosHashGet(pCxt->pSubTableHashObj, tbFName, len);
   if (NULL != pMeta) {
+    CHECK_CODE(ignoreAutoCreateTableClause(pCxt));
     return cloneTableMeta(*pMeta, &pCxt->pTableMeta);
   }
 
@@ -1522,13 +1542,6 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache
   return code;
 }
 
-typedef struct SInsertParseSyntaxCxt {
-  SParseContext*   pComCxt;
-  char*            pSql;
-  SMsgBuf          msg;
-  SParseMetaCache* pMetaCache;
-} SInsertParseSyntaxCxt;
-
 static int32_t skipParentheses(SInsertParseSyntaxCxt* pCxt) {
   SToken  sToken;
   int32_t expectRightParenthesis = 1;

source/libs/parser/src/parTokenizer.c

@@ -54,6 +54,7 @@ static SKeyword keywordTable[] = {
     {"CACHE",          TK_CACHE},
     {"CACHELAST",      TK_CACHELAST},
     {"CAST",           TK_CAST},
+    {"CLIENT_VERSION", TK_CLIENT_VERSION},
     {"CLUSTER",        TK_CLUSTER},
     {"COLUMN",         TK_COLUMN},
     {"COMMENT",        TK_COMMENT},
@@ -64,9 +65,10 @@ static SKeyword keywordTable[] = {
     {"CONNECTIONS",    TK_CONNECTIONS},
     {"CONSUMER",       TK_CONSUMER},
     {"CONSUMERS",      TK_CONSUMERS},
+    {"CONTAINS",       TK_CONTAINS},
     {"COUNT",          TK_COUNT},
     {"CREATE",         TK_CREATE},
-    {"CONTAINS",      TK_CONTAINS}, 
+    {"CURRENT_USER",   TK_CURRENT_USER},
     {"DATABASE",       TK_DATABASE},
     {"DATABASES",      TK_DATABASES},
     {"DBS",            TK_DBS},
@@ -169,6 +171,8 @@ static SKeyword keywordTable[] = {
     {"SCHEMALESS",     TK_SCHEMALESS},
     {"SCORES",         TK_SCORES},
     {"SELECT",         TK_SELECT},
+    {"SERVER_STATUS",  TK_SERVER_STATUS},
+    {"SERVER_VERSION", TK_SERVER_VERSION},
     {"SESSION",        TK_SESSION},
     {"SET",            TK_SET},
     {"SHOW",           TK_SHOW},

source/libs/parser/test/parInsertTest.cpp

@@ -265,6 +265,11 @@ TEST_F(InsertTest, autoCreateTableTest) {
       "insert into st1s1 using st1 (tag1, tag2) tags(1, 'wxy') values (now, 1, \"beijing\")"
       "(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")");
   ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS);
+
+  bind(
+      "insert into st1s1 using st1 tags(1, 'wxy', now) values (now, 1, \"beijing\")"
+      "st1s1 using st1 tags(1, 'wxy', now) values (now+1s, 2, \"shanghai\")");
+  ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
 }
 
 TEST_F(InsertTest, toleranceTest) {

source/libs/parser/test/parSelectTest.cpp

@@ -423,6 +423,18 @@ TEST_F(ParserSelectTest, withoutFrom) {
   useDb("root", "test");
 
   run("SELECT 1");
+
+  run("SELECT DATABASE()");
+
+  run("SELECT CLIENT_VERSION()");
+
+  run("SELECT SERVER_VERSION()");
+
+  run("SELECT SERVER_STATUS()");
+
+  run("SELECT CURRENT_USER()");
+
+  run("SELECT USER()");
 }
 
 }  // namespace ParserTest

source/libs/parser/test/parTestUtil.cpp

@@ -203,6 +203,7 @@ class ParserTestBaseImpl {
     pCxt->pMsg = stmtEnv_.msgBuf_.data();
     pCxt->msgLen = stmtEnv_.msgBuf_.max_size();
     pCxt->async = async;
+    pCxt->svrVer = "3.0.0.0";
   }
 
   void doParse(SParseContext* pCxt, SQuery** pQuery) {

source/libs/planner/src/planOptimizer.c

@@ -276,7 +276,8 @@ static int32_t pushDownCondOptAppendCond(SNode** pCond, SNode** pAdditionalCond)
   }
 
   int32_t code = TSDB_CODE_SUCCESS;
-  if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pCond)) {
+  if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pCond) &&
+      LOGIC_COND_TYPE_AND == ((SLogicConditionNode*)*pCond)->condType) {
     code = nodesListAppend(((SLogicConditionNode*)*pCond)->pParameterList, *pAdditionalCond);
     if (TSDB_CODE_SUCCESS == code) {
       *pAdditionalCond = NULL;
@@ -1083,6 +1084,11 @@ static int32_t partTagsOptRebuildTbanme(SNodeList* pPartKeys) {
   return code;
 }
 
+// todo refact: just to mask compilation warnings
+static void partTagsSetAlias(char* pAlias, int32_t len, const char* pTableAlias, const char* pColName) {
+  snprintf(pAlias, len, "%s.%s", pTableAlias, pColName);
+}
+
 static SNode* partTagsCreateWrapperFunc(const char* pFuncName, SNode* pNode) {
   SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
   if (NULL == pFunc) {
@@ -1092,7 +1098,7 @@ static SNode* partTagsCreateWrapperFunc(const char* pFuncName, SNode* pNode) {
   strcpy(pFunc->functionName, pFuncName);
   if (QUERY_NODE_COLUMN == nodeType(pNode)) {
     SColumnNode* pCol = (SColumnNode*)pNode;
-    sprintf(pFunc->node.aliasName, "%s.%s", pCol->tableAlias, pCol->colName);
+    partTagsSetAlias(pFunc->node.aliasName, sizeof(pFunc->node.aliasName), pCol->tableAlias, pCol->colName);
   } else {
     strcpy(pFunc->node.aliasName, ((SExprNode*)pNode)->aliasName);
   }
@@ -1464,9 +1470,9 @@ static SNode* rewriteUniqueOptCreateFirstFunc(SFunctionNode* pSelectValue, SNode
 
   strcpy(pFunc->functionName, "first");
   if (NULL != pSelectValue) {
-    sprintf(pFunc->node.aliasName, "%s", pSelectValue->node.aliasName);
+    strcpy(pFunc->node.aliasName, pSelectValue->node.aliasName);
   } else {
-    sprintf(pFunc->node.aliasName, "%s.%p", pFunc->functionName, pFunc);
+    snprintf(pFunc->node.aliasName, sizeof(pFunc->node.aliasName), "%s.%p", pFunc->functionName, pFunc);
   }
   int32_t code = nodesListMakeStrictAppend(&pFunc->pParameterList, nodesCloneNode(pCol));
   if (TSDB_CODE_SUCCESS == code) {

source/libs/planner/src/planner.c

@@ -85,11 +85,23 @@ 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;
+static void clearSubplanExecutionNode(SPhysiNode* pNode) {
+  if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pNode)) {
+    SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pNode;
+    NODES_DESTORY_LIST(pExchange->pSrcEndPoints);
+  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == nodeType(pNode)) {
+    SMergePhysiNode* pMerge = (SMergePhysiNode*)pNode;
+    pMerge->numOfChannels = LIST_LENGTH(pMerge->node.pChildren);
+    SNode* pChild = NULL;
+    FOREACH(pChild, pMerge->node.pChildren) { NODES_DESTORY_LIST(((SExchangePhysiNode*)pChild)->pSrcEndPoints); }
+  }
+
+  SNode* pChild = NULL;
+  FOREACH(pChild, pNode->pChildren) { clearSubplanExecutionNode((SPhysiNode*)pChild); }
 }
 
+void qClearSubplanExecutionNode(SSubplan* pSubplan) { clearSubplanExecutionNode(pSubplan->pNode); }
+
 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/planTestUtil.cpp

@@ -310,6 +310,7 @@ class PlannerTestBaseImpl {
     cxt.sqlLen = stmtEnv_.sql_.length();
     cxt.pMsg = stmtEnv_.msgBuf_.data();
     cxt.msgLen = stmtEnv_.msgBuf_.max_size();
+    cxt.svrVer = "3.0.0.0";
 
     DO_WITH_THROW(qParseSql, &cxt, pQuery);
     if (prepare) {

source/libs/sync/src/syncRaftLog.c

@@ -17,24 +17,27 @@
 #include "syncRaftCfg.h"
 #include "syncRaftStore.h"
 
-// refactor, log[0 .. n] ==> log[m .. n]
+//-------------------------------
+// log[m .. n]
+
+// public function
 static int32_t   raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
-// static int32_t   raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex);
 static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore);
 static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore);
 static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore);
 static bool      raftLogIsEmpty(struct SSyncLogStore* pLogStore);
 static int32_t   raftLogEntryCount(struct SSyncLogStore* pLogStore);
-
 static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore);
 static SyncTerm  raftLogLastTerm(struct SSyncLogStore* pLogStore);
 static int32_t   raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
 static int32_t   raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry);
 static int32_t   raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
 
+// private function
 static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);
 
 //-------------------------------
+// log[0 .. n]
 static SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore);
 static SyncIndex       logStoreLastIndex(SSyncLogStore* pLogStore);
 static SyncTerm        logStoreLastTerm(SSyncLogStore* pLogStore);
@@ -44,28 +47,86 @@ static int32_t         logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex from
 static int32_t         logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
 static SyncIndex       logStoreGetCommitIndex(SSyncLogStore* pLogStore);
 
-// refactor, log[0 .. n] ==> log[m .. n]
-/*
-static int32_t raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex) {
-  // if beginIndex == 0, donot need call this funciton
-  ASSERT(beginIndex > 0);
+//-------------------------------
+SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
+  SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
+  ASSERT(pLogStore != NULL);
+
+  pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
+  ASSERT(pLogStore->data != NULL);
 
   SSyncLogStoreData* pData = pLogStore->data;
-  SWal*              pWal = pData->pWal;
-  sTrace("vgId:%d, reset wal begin index:%ld", pData->pSyncNode->vgId, beginIndex);
+  pData->pSyncNode = pSyncNode;
+  pData->pWal = pSyncNode->pWal;
+  ASSERT(pData->pWal != NULL);
 
-  pData->beginIndex = beginIndex;
-  walRestoreFromSnapshot(pWal, beginIndex - 1);
-  return 0;
+  taosThreadMutexInit(&(pData->mutex), NULL);
+  pData->pWalHandle = walOpenReadHandle(pData->pWal);
+  ASSERT(pData->pWalHandle != NULL);
+
+  pLogStore->appendEntry = logStoreAppendEntry;
+  pLogStore->getEntry = logStoreGetEntry;
+  pLogStore->truncate = logStoreTruncate;
+  pLogStore->getLastIndex = logStoreLastIndex;
+  pLogStore->getLastTerm = logStoreLastTerm;
+  pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
+  pLogStore->getCommitIndex = logStoreGetCommitIndex;
+
+  pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
+  pLogStore->syncLogBeginIndex = raftLogBeginIndex;
+  pLogStore->syncLogEndIndex = raftLogEndIndex;
+  pLogStore->syncLogIsEmpty = raftLogIsEmpty;
+  pLogStore->syncLogEntryCount = raftLogEntryCount;
+  pLogStore->syncLogLastIndex = raftLogLastIndex;
+  pLogStore->syncLogLastTerm = raftLogLastTerm;
+  pLogStore->syncLogAppendEntry = raftLogAppendEntry;
+  pLogStore->syncLogGetEntry = raftLogGetEntry;
+  pLogStore->syncLogTruncate = raftLogTruncate;
+  pLogStore->syncLogWriteIndex = raftLogWriteIndex;
+
+  return pLogStore;
 }
-*/
 
+void logStoreDestory(SSyncLogStore* pLogStore) {
+  if (pLogStore != NULL) {
+    SSyncLogStoreData* pData = pLogStore->data;
+
+    taosThreadMutexLock(&(pData->mutex));
+    if (pData->pWalHandle != NULL) {
+      walCloseReadHandle(pData->pWalHandle);
+      pData->pWalHandle = NULL;
+    }
+    taosThreadMutexUnlock(&(pData->mutex));
+    taosThreadMutexDestroy(&(pData->mutex));
+
+    taosMemoryFree(pLogStore->data);
+    taosMemoryFree(pLogStore);
+  }
+}
+
+//-------------------------------
+// log[m .. n]
 static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
   ASSERT(snapshotIndex >= 0);
 
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
-  walRestoreFromSnapshot(pWal, snapshotIndex);
+  int32_t            code = walRestoreFromSnapshot(pWal, snapshotIndex);
+  if (code != 0) {
+    int32_t     err = terrno;
+    const char* errStr = tstrerror(err);
+    int32_t     sysErr = errno;
+    const char* sysErrStr = strerror(errno);
+
+    char logBuf[128];
+    snprintf(logBuf, sizeof(logBuf),
+             "wal restore from snapshot error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex, err,
+             err, errStr, sysErr, sysErrStr);
+    syncNodeErrorLog(pData->pSyncNode, logBuf);
+
+    return -1;
+  }
+
   return 0;
 }
 
@@ -91,18 +152,6 @@ static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {
   return count > 0 ? count : 0;
 }
 
-#if 0
-static bool raftLogInRange(struct SSyncLogStore* pLogStore, SyncIndex index) {
-  SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
-  SyncIndex endIndex = raftLogEndIndex(pLogStore);
-  if (index >= beginIndex && index <= endIndex) {
-    return true;
-  } else {
-    return false;
-  }
-}
-#endif
-
 static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {
   SyncIndex          lastIndex;
   SSyncLogStoreData* pData = pLogStore->data;
@@ -138,24 +187,6 @@ static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
   return 0;
 }
 
-/*
-static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
-  SyncTerm lastTerm = 0;
-  if (raftLogEntryCount(pLogStore) == 0) {
-    lastTerm = 0;
-  } else {
-    SSyncRaftEntry* pLastEntry;
-    int32_t         code = raftLogGetLastEntry(pLogStore, &pLastEntry);
-    ASSERT(code == 0);
-    if (pLastEntry != NULL) {
-      lastTerm = pLastEntry->term;
-      taosMemoryFree(pLastEntry);
-    }
-  }
-  return lastTerm;
-}
-*/
-
 static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
@@ -197,53 +228,6 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
   return code;
 }
 
-#if 0
-static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
-  SSyncLogStoreData* pData = pLogStore->data;
-  SWal*              pWal = pData->pWal;
-  int32_t            code;
-
-  *ppEntry = NULL;
-  if (raftLogInRange(pLogStore, index)) {
-    SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
-    ASSERT(pWalHandle != NULL);
-
-    code = walReadWithHandle(pWalHandle, index);
-    if (code != 0) {
-      int32_t     err = terrno;
-      const char* errStr = tstrerror(err);
-      int32_t     linuxErr = errno;
-      const char* linuxErrMsg = strerror(errno);
-      sError("raftLogGetEntry error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr,
-             linuxErrMsg);
-      ASSERT(0);
-      walCloseReadHandle(pWalHandle);
-      return code;
-    }
-
-    *ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
-    ASSERT(*ppEntry != NULL);
-    (*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;
-    (*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;
-    (*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
-    (*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
-    (*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;
-    (*ppEntry)->index = index;
-    ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
-    memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
-
-    // need to hold, do not new every time!!
-    walCloseReadHandle(pWalHandle);
-
-  } else {
-    // index not in range
-    code = 0;
-  }
-
-  return code;
-}
-#endif
-
 static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
@@ -343,91 +327,8 @@ static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** pp
   return -1;
 }
 
-/*
-static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {
-  *ppLastEntry = NULL;
-  if (raftLogEntryCount(pLogStore) == 0) {
-    return 0;
-  }
-  SyncIndex lastIndex = raftLogLastIndex(pLogStore);
-  int32_t   code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);
-  return code;
-}
-*/
-
-//-------------------------------
-SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
-  SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
-  ASSERT(pLogStore != NULL);
-
-  pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
-  ASSERT(pLogStore->data != NULL);
-
-  SSyncLogStoreData* pData = pLogStore->data;
-  pData->pSyncNode = pSyncNode;
-  pData->pWal = pSyncNode->pWal;
-  ASSERT(pData->pWal != NULL);
-
-  taosThreadMutexInit(&(pData->mutex), NULL);
-  pData->pWalHandle = walOpenReadHandle(pData->pWal);
-  ASSERT(pData->pWalHandle != NULL);
-
-  /*
-    SyncIndex firstVer = walGetFirstVer(pData->pWal);
-    SyncIndex lastVer = walGetLastVer(pData->pWal);
-    if (firstVer >= 0) {
-      pData->beginIndex = firstVer;
-    } else if (firstVer == -1) {
-      pData->beginIndex = lastVer + 1;
-    } else {
-      ASSERT(0);
-    }
-  */
-
-  pLogStore->appendEntry = logStoreAppendEntry;
-  pLogStore->getEntry = logStoreGetEntry;
-  pLogStore->truncate = logStoreTruncate;
-  pLogStore->getLastIndex = logStoreLastIndex;
-  pLogStore->getLastTerm = logStoreLastTerm;
-  pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
-  pLogStore->getCommitIndex = logStoreGetCommitIndex;
-
-  // pLogStore->syncLogSetBeginIndex = raftLogSetBeginIndex;
-  pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
-  pLogStore->syncLogBeginIndex = raftLogBeginIndex;
-  pLogStore->syncLogEndIndex = raftLogEndIndex;
-  pLogStore->syncLogIsEmpty = raftLogIsEmpty;
-  pLogStore->syncLogEntryCount = raftLogEntryCount;
-  pLogStore->syncLogLastIndex = raftLogLastIndex;
-  pLogStore->syncLogLastTerm = raftLogLastTerm;
-  pLogStore->syncLogAppendEntry = raftLogAppendEntry;
-  pLogStore->syncLogGetEntry = raftLogGetEntry;
-  pLogStore->syncLogTruncate = raftLogTruncate;
-  pLogStore->syncLogWriteIndex = raftLogWriteIndex;
-
-  // pLogStore->syncLogInRange = raftLogInRange;
-
-  return pLogStore;
-}
-
-void logStoreDestory(SSyncLogStore* pLogStore) {
-  if (pLogStore != NULL) {
-    SSyncLogStoreData* pData = pLogStore->data;
-
-    taosThreadMutexLock(&(pData->mutex));
-    if (pData->pWalHandle != NULL) {
-      walCloseReadHandle(pData->pWalHandle);
-      pData->pWalHandle = NULL;
-    }
-    taosThreadMutexUnlock(&(pData->mutex));
-    taosThreadMutexDestroy(&(pData->mutex));
-
-    taosMemoryFree(pLogStore->data);
-    taosMemoryFree(pLogStore);
-  }
-}
-
 //-------------------------------
+// log[0 .. n]
 int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
@@ -593,55 +494,6 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
   return pEntry;
 }
 
-/*
-cJSON* logStore2Json(SSyncLogStore* pLogStore) {
-  char               u64buf[128] = {0};
-  SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
-  cJSON*             pRoot = cJSON_CreateObject();
-
-  if (pData != NULL && pData->pWal != NULL) {
-    snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
-    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
-    snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
-    cJSON_AddStringToObject(pRoot, "pWal", u64buf);
-
-    snprintf(u64buf, sizeof(u64buf), "%ld", pData->beginIndex);
-    cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);
-
-    SyncIndex endIndex = raftLogEndIndex(pLogStore);
-    snprintf(u64buf, sizeof(u64buf), "%ld", endIndex);
-    cJSON_AddStringToObject(pRoot, "endIndex", u64buf);
-
-    int32_t count = raftLogEntryCount(pLogStore);
-    cJSON_AddNumberToObject(pRoot, "entryCount", count);
-
-    snprintf(u64buf, sizeof(u64buf), "%ld", raftLogWriteIndex(pLogStore));
-    cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);
-
-    snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
-    cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);
-
-    snprintf(u64buf, sizeof(u64buf), "%ld", raftLogLastIndex(pLogStore));
-    cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
-    snprintf(u64buf, sizeof(u64buf), "%lu", raftLogLastTerm(pLogStore));
-    cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
-
-    cJSON* pEntries = cJSON_CreateArray();
-    cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
-
-    for (SyncIndex i = pData->beginIndex; i <= endIndex; ++i) {
-      SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
-      cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
-      syncEntryDestory(pEntry);
-    }
-  }
-
-  cJSON* pJson = cJSON_CreateObject();
-  cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);
-  return pJson;
-}
-*/
-
 cJSON* logStore2Json(SSyncLogStore* pLogStore) {
   char               u64buf[128] = {0};
   SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;

source/libs/sync/src/syncSnapshot.c

@@ -508,31 +508,50 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
   } while (0);
 }
 
-static void snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
+static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
   ASSERT(pMsg->seq == SYNC_SNAPSHOT_SEQ_END);
 
-  if (pReceiver->pWriter != NULL) {
   int32_t code = 0;
+  if (pReceiver->pWriter != NULL) {
+    // write data
     if (pMsg->dataLen > 0) {
       code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data,
                                                            pMsg->dataLen);
-      ASSERT(code == 0);
+      if (code != 0) {
+        syncNodeErrorLog(pReceiver->pSyncNode, "snapshot write error");
+        return -1;
       }
-
-    code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true);
-    ASSERT(code == 0);
-    pReceiver->pWriter = NULL;
     }
 
-  pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
+    // reset wal
+    code =
+        pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
+    if (code != 0) {
+      syncNodeErrorLog(pReceiver->pSyncNode, "wal restore from snapshot error");
+      return -1;
+    }
 
     // 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);
+    // stop writer
+    code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true);
+    if (code != 0) {
+      syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
+      ASSERT(0);
+      return -1;
+    }
+    pReceiver->pWriter = NULL;
+
+    // update progress
+    pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
+
+  } else {
+    syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
+    return -1;
+  }
 
   // event log
   do {
@@ -542,6 +561,8 @@ static void snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapsho
     syncNodeEventLog(pReceiver->pSyncNode, eventLog);
     taosMemoryFree(eventLog);
   } while (0);
+
+  return 0;
 }
 
 static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
@@ -642,6 +663,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
   // get receiver
   SSyncSnapshotReceiver *pReceiver = pSyncNode->pNewNodeReceiver;
   bool                   needRsp = false;
+  int32_t                code = 0;
 
   // state, term, seq/ack
   if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
@@ -653,8 +675,10 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
 
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) {
         // end, finish FSM
-        snapshotReceiverFinish(pReceiver, pMsg);
+        code = snapshotReceiverFinish(pReceiver, pMsg);
+        if (code == 0) {
           snapshotReceiverStop(pReceiver);
+        }
         needRsp = true;
 
         // maybe update lastconfig
@@ -772,6 +796,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
         snapshotReSend(pSender);
 
       } else {
+        // error log
         do {
           char logBuf[96];
           snprintf(logBuf, sizeof(logBuf), "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);

source/libs/transport/inc/transComm.h

@@ -148,6 +148,7 @@ typedef struct {
   char release : 2;
   char secured : 2;
   char spi : 2;
+  char hasEpSet : 2;  // contain epset or not, 0(default): no epset, 1: contain epset
 
   char     user[TSDB_UNI_LEN];
   STraceId traceId;
@@ -378,12 +379,9 @@ typedef struct SDelayQueue {
 } SDelayQueue;
 
 int  transDQCreate(uv_loop_t* loop, SDelayQueue** queue);
-
 void transDQDestroy(SDelayQueue* queue);
-
 int  transDQSched(SDelayQueue* queue, void (*func)(void* arg), void* arg, uint64_t timeoutMs);
 
-// void transPrintEpSet(SEpSet* pEpSet);
 bool transEpSetIsEqual(SEpSet* a, SEpSet* b);
 /*
  * init global func

source/libs/transport/src/trans.c

@@ -141,8 +141,10 @@ void rpcSendRecv(void* shandle, SEpSet* pEpSet, SRpcMsg* pMsg, SRpcMsg* pRsp) {
 }
 
 void    rpcSendResponse(const SRpcMsg* pMsg) { transSendResponse(pMsg); }
+
 int32_t rpcGetConnInfo(void* thandle, SRpcConnInfo* pInfo) { return 0; }
 
+
 void rpcRefHandle(void* handle, int8_t type) {
   assert(type == TAOS_CONN_SERVER || type == TAOS_CONN_CLIENT);
   (*taosRefHandle[type])(handle);

source/libs/transport/src/transCli.c

@@ -309,6 +309,7 @@ void cliHandleResp(SCliConn* conn) {
   transMsg.msgType = pHead->msgType;
   transMsg.info.ahandle = NULL;
   transMsg.info.traceId = pHead->traceId;
+  transMsg.info.hasEpSet = pHead->hasEpSet;
 
   SCliMsg*       pMsg = NULL;
   STransConnCtx* pCtx = NULL;
@@ -1020,6 +1021,23 @@ void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) {
     *val = newVal;
   }
 }
+
+bool cliTryToExtractEpSet(STransMsg* pResp, SEpSet* dst) {
+  if (pResp == NULL || pResp->info.hasEpSet == 0) {
+    return false;
+  }
+  tDeserializeSEpSet(pResp->pCont, pResp->contLen, dst);
+  int32_t tlen = tSerializeSEpSet(NULL, 0, dst);
+
+  int32_t bufLen = pResp->contLen - tlen;
+  char*   buf = rpcMallocCont(bufLen);
+
+  memcpy(buf, (char*)pResp->pCont + tlen, bufLen);
+
+  pResp->pCont = buf;
+  pResp->contLen = bufLen;
+  return true;
+}
 int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
   SCliThrd* pThrd = pConn->hostThrd;
   STrans*   pTransInst = pThrd->pTransInst;
@@ -1064,6 +1082,12 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
   }
 
   STraceId* trace = &pResp->info.traceId;
+
+  if (cliTryToExtractEpSet(pResp, &pCtx->epSet)) {
+    char tbuf[256] = {0};
+    EPSET_DEBUG_STR(&pCtx->epSet, tbuf);
+    tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn);
+  }
   if (pCtx->pSem != NULL) {
     tGTrace("%s conn %p(sync) handle resp", CONN_GET_INST_LABEL(pConn), pConn);
     if (pCtx->pRsp == NULL) {

source/libs/transport/src/transSvr.c

@@ -140,6 +140,7 @@ static void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd);
 static void (*transAsyncHandle[])(SSvrMsg* msg, SWorkThrd* thrd) = {uvHandleResp, uvHandleQuit, uvHandleRelease,
                                                                     uvHandleRegister, NULL};
 
+
 static void uvDestroyConn(uv_handle_t* handle);
 
 // server and worker thread
@@ -337,7 +338,7 @@ void uvOnSendCb(uv_write_t* req, int status) {
     if (!transQueueEmpty(&conn->srvMsgs)) {
       SSvrMsg* msg = transQueuePop(&conn->srvMsgs);
       destroySmsg(msg);
-      // send second data, just use for push
+      // send cached data
       if (!transQueueEmpty(&conn->srvMsgs)) {
         msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
         if (msg->type == Register && conn->status == ConnAcquire) {
@@ -375,7 +376,6 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) {
     tError("fail to dispatch conn to work thread");
   }
   uv_close((uv_handle_t*)req->data, uvFreeCb);
-  // taosMemoryFree(req->data);
   taosMemoryFree(req);
 }
 
@@ -389,6 +389,7 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
   STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
   pHead->ahandle = (uint64_t)pMsg->info.ahandle;
   pHead->traceId = pMsg->info.traceId;
+  pHead->hasEpSet = pMsg->info.hasEpSet;
 
   if (pConn->status == ConnNormal) {
     pHead->msgType = pConn->inType + 1;
@@ -401,6 +402,7 @@ static void uvPrepareSendData(SSvrMsg* smsg, uv_buf_t* wb) {
       transUnrefSrvHandle(pConn);
     } else {
       pHead->msgType = pMsg->msgType;
+      // set up resp msg type
       if (pHead->msgType == 0 && transMsgLenFromCont(pMsg->contLen) == sizeof(STransMsgHead))
         pHead->msgType = pConn->inType + 1;
     }

source/os/src/osSocket.c

@@ -925,10 +925,24 @@ uint32_t taosGetIpv4FromFqdn(const char *fqdn) {
 }
 
 int32_t taosGetFqdn(char *fqdn) {
+#ifdef WINDOWS
+  // Initialize Winsock
+  WSADATA wsaData;
+  int     iResult;
+  iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
+  if (iResult != 0) {
+    // printf("WSAStartup failed: %d\n", iResult);
+    return 1;
+  }
+#endif
   char hostname[1024];
   hostname[1023] = '\0';
   if (gethostname(hostname, 1023) == -1) {
-    // printf("failed to get hostname, reason:%s", strerror(errno));
+#ifdef WINDOWS
+    printf("failed to get hostname, reason:%s", strerror(WSAGetLastError()));
+#else
+    printf("failed to get hostname, reason:%s", strerror(errno));
+#endif
     assert(0);
     return -1;
   }

tests/pytest/a.sh

+#!/bin/bash
+for i in {1..100}
+do 
+	echo $i
+        python3 ./test.py -f query/nestedQuery/nestedQuery_datacheck.py >>log 2>&1
+        if [ $? -eq 0 ] 
+	then
+		echo success
+	else
+		echo failed
+		break
+	fi
+done

tests/script/sh/exec.bat

@@ -6,6 +6,7 @@ if %1 == -n set NODE_NAME=%2
 if %1 == -s set EXEC_OPTON=%2
 if %3 == -n set NODE_NAME=%4
 if %3 == -s set EXEC_OPTON=%4
+if "%5" == "-x" set EXEC_SIGNAL=%6
 
 rem echo NODE_NAME:  %NODE_NAME%
 rem echo NODE:       %EXEC_OPTON%
@@ -39,7 +40,7 @@ rem echo TAOS_LOG:   %TAOS_LOG%
 if %EXEC_OPTON% == start (
   rm -rf %TAOS_LOG%
   echo start %TAOSD% -c %CFG_DIR%
-  start %TAOSD% -c %CFG_DIR%
+  mintty -h never %TAOSD% -c %CFG_DIR%
   set /a check_num=0
 :check_online
   sleep 1
@@ -58,13 +59,25 @@ if %EXEC_OPTON% == stop (
   rem echo wmic process where "name='taosd.exe' and CommandLine like '%%%NODE_NAME%%%'" list INSTANCE
   rem wmic process where "name='taosd.exe' and CommandLine like '%%%NODE_NAME%%%'" call terminate > NUL 2>&1
   
-  for /f "tokens=1 skip=1" %%A in (
-    'wmic process where "name='taosd.exe' and CommandLine like '%%%NODE_NAME%%%'" get processId '
-  ) do (
-    rem echo taskkill /IM %%A 
-    taskkill /IM %%A > NUL 2>&1
+  for /f "tokens=2" %%A in ('wmic process where "name='taosd.exe' and CommandLine like '%%%NODE_NAME%%%'" get processId ^| xargs echo') do (
+    for /f "tokens=1" %%B in ('ps ^| grep %%A') do (
+      if "%EXEC_SIGNAL%" == "SIGKILL" (
+        kill -9 %%B
+      ) else (
+        kill -INT %%B
+        call :check_offline
+      )
+    )
     goto :finish
   ) 
 )
-
 :finish
+goto :eof
+
+:check_offline
+sleep 1
+for /f "tokens=2" %%C in ('wmic process where "name='taosd.exe' and CommandLine like '%%%NODE_NAME%%%'" get processId ^| xargs echo') do (
+  echo check taosd offline
+  goto :check_offline
+)
+goto :eof
\ No newline at end of file

tests/script/test-all.bat

@@ -62,7 +62,5 @@ goto :eof
 
 :CheckSkipCase
 set skipCase=false
-if "%*" == "./test.sh -f tsim/query/scalarFunction.sim" ( set skipCase=true )
-if "%*" == "./test.sh -f tsim/stream/distributeInterval0.sim" ( set skipCase=true )
-if "%*" == "./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim" ( set skipCase=true )
+@REM if "%*" == "./test.sh -f tsim/query/scalarFunction.sim" ( set skipCase=true )
 :goto eof
\ No newline at end of file

tests/system-test/fulltest.sh

@@ -23,7 +23,7 @@ python3 ./test.py -f 1-insert/alter_stable.py
 python3 ./test.py -f 1-insert/alter_table.py
 python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
 python3 ./test.py -f 1-insert/table_comment.py
-python3 ./test.py -f 1-insert/table_param_ttl.py
+#python3 ./test.py -f 1-insert/table_param_ttl.py
 python3 ./test.py -f 2-query/between.py
 python3 ./test.py -f 2-query/distinct.py
 python3 ./test.py -f 2-query/varchar.py

taos-tools @ 71050276

-Subproject commit 28a49b447f71c4f014ebbac858b7215b897d57fd
+Subproject commit 7105027650b51e701cfa1dac11b8fb42d447dd01