Merge pull request #13749 from taosdata/feature/3.0_wxy

fix: handle the memory leak of parser

include/common/tmsg.h

@@ -1506,6 +1506,7 @@ typedef struct {
   char*   sql;
   char*   ast;
   int8_t  triggerType;
+  int64_t maxDelay;
   int64_t watermark;
 } SCMCreateStreamReq;
 

include/common/ttokendef.h

@@ -184,76 +184,77 @@
 #define TK_TRIGGER       166
 #define TK_AT_ONCE       167
 #define TK_WINDOW_CLOSE  168
-#define TK_WATERMARK     169
-#define TK_KILL          170
-#define TK_CONNECTION    171
-#define TK_TRANSACTION   172
-#define TK_BALANCE       173
-#define TK_VGROUP        174
-#define TK_MERGE         175
-#define TK_REDISTRIBUTE  176
-#define TK_SPLIT         177
-#define TK_SYNCDB        178
-#define TK_DELETE        179
-#define TK_NULL          180
-#define TK_NK_QUESTION   181
-#define TK_NK_ARROW      182
-#define TK_ROWTS         183
-#define TK_TBNAME        184
-#define TK_QSTARTTS      185
-#define TK_QENDTS        186
-#define TK_WSTARTTS      187
-#define TK_WENDTS        188
-#define TK_WDURATION     189
-#define TK_CAST          190
-#define TK_NOW           191
-#define TK_TODAY         192
-#define TK_TIMEZONE      193
-#define TK_COUNT         194
-#define TK_FIRST         195
-#define TK_LAST          196
-#define TK_LAST_ROW      197
-#define TK_BETWEEN       198
-#define TK_IS            199
-#define TK_NK_LT         200
-#define TK_NK_GT         201
-#define TK_NK_LE         202
-#define TK_NK_GE         203
-#define TK_NK_NE         204
-#define TK_MATCH         205
-#define TK_NMATCH        206
-#define TK_CONTAINS      207
-#define TK_JOIN          208
-#define TK_INNER         209
-#define TK_SELECT        210
-#define TK_DISTINCT      211
-#define TK_WHERE         212
-#define TK_PARTITION     213
-#define TK_BY            214
-#define TK_SESSION       215
-#define TK_STATE_WINDOW  216
-#define TK_SLIDING       217
-#define TK_FILL          218
-#define TK_VALUE         219
-#define TK_NONE          220
-#define TK_PREV          221
-#define TK_LINEAR        222
-#define TK_NEXT          223
-#define TK_HAVING        224
-#define TK_ORDER         225
-#define TK_SLIMIT        226
-#define TK_SOFFSET       227
-#define TK_LIMIT         228
-#define TK_OFFSET        229
-#define TK_ASC           230
-#define TK_NULLS         231
-#define TK_ID            232
-#define TK_NK_BITNOT     233
-#define TK_INSERT        234
-#define TK_VALUES        235
-#define TK_IMPORT        236
-#define TK_NK_SEMI       237
-#define TK_FILE          238
+#define TK_MAX_DELAY     169
+#define TK_WATERMARK     170
+#define TK_KILL          171
+#define TK_CONNECTION    172
+#define TK_TRANSACTION   173
+#define TK_BALANCE       174
+#define TK_VGROUP        175
+#define TK_MERGE         176
+#define TK_REDISTRIBUTE  177
+#define TK_SPLIT         178
+#define TK_SYNCDB        179
+#define TK_DELETE        180
+#define TK_NULL          181
+#define TK_NK_QUESTION   182
+#define TK_NK_ARROW      183
+#define TK_ROWTS         184
+#define TK_TBNAME        185
+#define TK_QSTARTTS      186
+#define TK_QENDTS        187
+#define TK_WSTARTTS      188
+#define TK_WENDTS        189
+#define TK_WDURATION     190
+#define TK_CAST          191
+#define TK_NOW           192
+#define TK_TODAY         193
+#define TK_TIMEZONE      194
+#define TK_COUNT         195
+#define TK_FIRST         196
+#define TK_LAST          197
+#define TK_LAST_ROW      198
+#define TK_BETWEEN       199
+#define TK_IS            200
+#define TK_NK_LT         201
+#define TK_NK_GT         202
+#define TK_NK_LE         203
+#define TK_NK_GE         204
+#define TK_NK_NE         205
+#define TK_MATCH         206
+#define TK_NMATCH        207
+#define TK_CONTAINS      208
+#define TK_JOIN          209
+#define TK_INNER         210
+#define TK_SELECT        211
+#define TK_DISTINCT      212
+#define TK_WHERE         213
+#define TK_PARTITION     214
+#define TK_BY            215
+#define TK_SESSION       216
+#define TK_STATE_WINDOW  217
+#define TK_SLIDING       218
+#define TK_FILL          219
+#define TK_VALUE         220
+#define TK_NONE          221
+#define TK_PREV          222
+#define TK_LINEAR        223
+#define TK_NEXT          224
+#define TK_HAVING        225
+#define TK_ORDER         226
+#define TK_SLIMIT        227
+#define TK_SOFFSET       228
+#define TK_LIMIT         229
+#define TK_OFFSET        230
+#define TK_ASC           231
+#define TK_NULLS         232
+#define TK_ID            233
+#define TK_NK_BITNOT     234
+#define TK_INSERT        235
+#define TK_VALUES        236
+#define TK_IMPORT        237
+#define TK_NK_SEMI       238
+#define TK_FILE          239
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/function/functionMgt.h

@@ -175,6 +175,7 @@ bool fmIsRepeatScanFunc(int32_t funcId);
 bool fmIsUserDefinedFunc(int32_t funcId);
 bool fmIsDistExecFunc(int32_t funcId);
 bool fmIsForbidFillFunc(int32_t funcId);
+bool fmIsForbidStreamFunc(int32_t funcId);
 
 int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
 

include/libs/nodes/cmdnodes.h

@@ -289,6 +289,7 @@ typedef struct SKillStmt {
 typedef struct SStreamOptions {
   ENodeType type;
   int8_t    triggerType;
+  SNode*    pDelay;
   SNode*    pWatermark;
 } SStreamOptions;
 

include/libs/nodes/nodes.h

@@ -252,22 +252,20 @@ typedef struct SNodeList {
   SListCell* pTail;
 } SNodeList;
 
-#define SNodeptr void*
-
-SNodeptr nodesMakeNode(ENodeType type);
-void     nodesDestroyNode(SNodeptr pNode);
+SNode* nodesMakeNode(ENodeType type);
+void   nodesDestroyNode(SNode* pNode);
 
 SNodeList* nodesMakeList();
-int32_t    nodesListAppend(SNodeList* pList, SNodeptr pNode);
-int32_t    nodesListStrictAppend(SNodeList* pList, SNodeptr pNode);
-int32_t    nodesListMakeAppend(SNodeList** pList, SNodeptr pNode);
-int32_t    nodesListMakeStrictAppend(SNodeList** pList, SNodeptr pNode);
+int32_t    nodesListAppend(SNodeList* pList, SNode* pNode);
+int32_t    nodesListStrictAppend(SNodeList* pList, SNode* pNode);
+int32_t    nodesListMakeAppend(SNodeList** pList, SNode* pNode);
+int32_t    nodesListMakeStrictAppend(SNodeList** pList, SNode* pNode);
 int32_t    nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
 int32_t    nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc);
-int32_t    nodesListPushFront(SNodeList* pList, SNodeptr pNode);
+int32_t    nodesListPushFront(SNodeList* pList, SNode* pNode);
 SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
 void       nodesListInsertList(SNodeList* pTarget, SListCell* pPos, SNodeList* pSrc);
-SNodeptr   nodesListGetNode(SNodeList* pList, int32_t index);
+SNode*     nodesListGetNode(SNodeList* pList, int32_t index);
 void       nodesDestroyList(SNodeList* pList);
 // Only clear the linked list structure, without releasing the elements inside
 void nodesClearList(SNodeList* pList);
@@ -275,9 +273,9 @@ void nodesClearList(SNodeList* pList);
 typedef enum EDealRes { DEAL_RES_CONTINUE = 1, DEAL_RES_IGNORE_CHILD, DEAL_RES_ERROR, DEAL_RES_END } EDealRes;
 
 typedef EDealRes (*FNodeWalker)(SNode* pNode, void* pContext);
-void nodesWalkExpr(SNodeptr pNode, FNodeWalker walker, void* pContext);
+void nodesWalkExpr(SNode* pNode, FNodeWalker walker, void* pContext);
 void nodesWalkExprs(SNodeList* pList, FNodeWalker walker, void* pContext);
-void nodesWalkExprPostOrder(SNodeptr pNode, FNodeWalker walker, void* pContext);
+void nodesWalkExprPostOrder(SNode* pNode, FNodeWalker walker, void* pContext);
 void nodesWalkExprsPostOrder(SNodeList* pList, FNodeWalker walker, void* pContext);
 
 typedef EDealRes (*FNodeRewriter)(SNode** pNode, void* pContext);
@@ -286,13 +284,13 @@ void nodesRewriteExprs(SNodeList* pList, FNodeRewriter rewriter, void* pContext)
 void nodesRewriteExprPostOrder(SNode** pNode, FNodeRewriter rewriter, void* pContext);
 void nodesRewriteExprsPostOrder(SNodeList* pList, FNodeRewriter rewriter, void* pContext);
 
-bool nodesEqualNode(const SNodeptr a, const SNodeptr b);
+bool nodesEqualNode(const SNode* a, const SNode* b);
 
-SNodeptr   nodesCloneNode(const SNodeptr pNode);
+SNode*     nodesCloneNode(const SNode* pNode);
 SNodeList* nodesCloneList(const SNodeList* pList);
 
 const char* nodesNodeName(ENodeType type);
-int32_t     nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen);
+int32_t     nodesNodeToString(const SNode* pNode, bool format, char** pStr, int32_t* pLen);
 int32_t     nodesStringToNode(const char* pStr, SNode** pNode);
 
 int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int32_t* pLen);

include/libs/nodes/querynodes.h

@@ -351,7 +351,6 @@ typedef struct SQuery {
   int32_t        placeholderNum;
   SArray*        pPlaceholderValues;
   SNode*         pPrepareRoot;
-  struct SParseMetaCache* pMetaCache;
 } SQuery;
 
 void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);

source/client/src/clientHb.c

@@ -698,7 +698,7 @@ SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
     return NULL;
   }
 
-  taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
+  // taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
 
   taosThreadMutexLock(&clientHbMgr.lock);
   taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);

source/client/src/clientImpl.c

@@ -233,9 +233,7 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
   return TSDB_CODE_SUCCESS;
 }
 
-static SAppInstInfo* getAppInfo(SRequestObj* pRequest) {
-  return pRequest->pTscObj->pAppInfo;
-}
+static SAppInstInfo* getAppInfo(SRequestObj* pRequest) { return pRequest->pTscObj->pAppInfo; }
 
 void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
   SRetrieveTableRsp* pRsp = NULL;
@@ -259,7 +257,7 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
   }
 
   pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
-//  pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
+  //  pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
 }
 
 int32_t asyncExecDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
@@ -404,16 +402,14 @@ int32_t scheduleAsyncQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNod
   SQueryResult     res = {.code = 0, .numOfRows = 0};
   SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
                            .requestId = pRequest->requestId,
-                           .requestObjRefId = pRequest->self
-  };
+                           .requestObjRefId = pRequest->self};
   SSchedulerReq    req = {.pConn = &conn,
                           .pNodeList = pNodeList,
                           .pDag = pDag,
                           .sql = pRequest->sqlstr,
                           .startTs = pRequest->metric.start,
                           .fp = schdExecCallback,
-                       .cbParam = &res
-  };
+                          .cbParam = &res};
 
   int32_t code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
 
@@ -458,18 +454,16 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
   SQueryResult     res = {0};
   SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
                            .requestId = pRequest->requestId,
-                           .requestObjRefId = pRequest->self
-  };
+                           .requestObjRefId = pRequest->self};
   SSchedulerReq    req = {.pConn = &conn,
                           .pNodeList = pNodeList,
                           .pDag = pDag,
                           .sql = pRequest->sqlstr,
                           .startTs = pRequest->metric.start,
                           .fp = NULL,
-                       .cbParam = NULL
-  };
+                          .cbParam = NULL};
 
-  int32_t code = schedulerExecJob(&req,&pRequest->body.queryJob, &res);
+  int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res);
   pRequest->body.resInfo.execRes = res.res;
 
   if (code != TSDB_CODE_SUCCESS) {
@@ -482,7 +476,8 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
     return pRequest->code;
   }
 
-  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type || TDMT_VND_CREATE_TABLE == pRequest->type) {
+  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type ||
+      TDMT_VND_CREATE_TABLE == pRequest->type) {
     pRequest->body.resInfo.numOfRows = res.numOfRows;
 
     if (pRequest->body.queryJob != 0) {
@@ -495,7 +490,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
   return pRequest->code;
 }
 
-int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet *epset) {
+int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
   int32_t     code = 0;
   SArray*     pArray = NULL;
   SSubmitRsp* pRsp = (SSubmitRsp*)res;
@@ -532,7 +527,7 @@ _return:
   return code;
 }
 
-int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet *epset) {
+int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
   int32_t code = 0;
   SArray* pArray = NULL;
   SArray* pTbArray = (SArray*)res;
@@ -601,7 +596,7 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
       break;
     }
     default:
-      tscError("0x%"PRIx64", invalid exec result for request type %d, reqId:0x%"PRIx64, pRequest->self,
+      tscError("0x%" PRIx64 ", invalid exec result for request type %d, reqId:0x%" PRIx64, pRequest->self,
                pRequest->type, pRequest->requestId);
       code = TSDB_CODE_APP_ERROR;
   }
@@ -610,13 +605,13 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
 }
 
 void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) {
-  SRequestObj* pRequest = (SRequestObj*) param;
+  SRequestObj* pRequest = (SRequestObj*)param;
   pRequest->code = code;
 
   STscObj* pTscObj = pRequest->pTscObj;
   if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code)) {
-    tscDebug("0x%"PRIx64" client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code),
-        pRequest->retry, pRequest->requestId);
+    tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
+             pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
     pRequest->prevCode = code;
     doAsyncQuery(pRequest, true);
     return;
@@ -727,30 +722,29 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery) {
       if (TSDB_CODE_SUCCESS == code) {
         code = qCreateQueryPlan(&cxt, &pRequest->body.pDag, pNodeList);
         if (code) {
-          tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
+          tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
+                   pRequest->requestId);
         }
       }
 
       if (TSDB_CODE_SUCCESS == code) {
-        SRequestConnInfo conn = {.pTrans = pAppInfo->pTransporter, 
-                                 .requestId = pRequest->requestId,
-                                 .requestObjRefId = pRequest->self
-        };
+        SRequestConnInfo conn = {
+            .pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
         SSchedulerReq req = {.pConn = &conn,
                              .pNodeList = pNodeList,
                              .pDag = pRequest->body.pDag,
                              .sql = pRequest->sqlstr,
                              .startTs = pRequest->metric.start,
                              .fp = schedulerExecCb,
-                             .cbParam = pRequest
-        };      
+                             .cbParam = pRequest};
         code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
       } else {
-        tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
+        tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
+                 pRequest->requestId);
         pRequest->body.queryFp(pRequest->body.param, pRequest, code);
       }
 
-      //todo not to be released here
+      // todo not to be released here
       taosArrayDestroy(pNodeList);
       break;
     }
@@ -997,9 +991,8 @@ void updateTargetEpSet(SMsgSendInfo* pSendInfo, STscObj* pTscObj, SRpcMsg* pMsg,
       SEpSet* pOrig = &pTscObj->pAppInfo->mgmtEp.epSet;
       SEp*    pOrigEp = &pOrig->eps[pOrig->inUse];
       SEp*    pNewEp = &pEpSet->eps[pEpSet->inUse];
-      tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in client", 
-          pOrig->inUse, pOrig->numOfEps, pOrigEp->fqdn, pOrigEp->port, 
-          pEpSet->inUse, pEpSet->numOfEps, pNewEp->fqdn, pNewEp->port);
+      tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in client", pOrig->inUse, pOrig->numOfEps,
+               pOrigEp->fqdn, pOrigEp->port, pEpSet->inUse, pEpSet->numOfEps, pNewEp->fqdn, pNewEp->port);
       updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, pEpSet);
       break;
     case TARGET_TYPE_VNODE: {
@@ -1415,14 +1408,14 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
   p += sizeof(uint64_t);
 
   // check fields
-  for(int32_t i = 0; i < numOfCols; ++i) {
-    int16_t type = *(int16_t*) p;
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    int16_t type = *(int16_t*)p;
     p += sizeof(int16_t);
 
-    int32_t bytes = *(int32_t*) p;
+    int32_t bytes = *(int32_t*)p;
     p += sizeof(int32_t);
 
-//    ASSERT(type == pFields[i].type && bytes == pFields[i].bytes);
+    //    ASSERT(type == pFields[i].type && bytes == pFields[i].bytes);
   }
 
   int32_t* colLength = (int32_t*)p;

source/client/src/clientMain.c

@@ -30,7 +30,7 @@
 #define TSC_VAR_RELEASED    0
 
 static int32_t sentinel = TSC_VAR_NOT_RELEASE;
-static int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt);
+static int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt);
 
 int taos_options(TSDB_OPTION option, const void *arg, ...) {
   static int32_t lock = 0;
@@ -177,8 +177,8 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
   return pResInfo->userFields;
 }
 
-static void syncQueryFn(void* param, void* res, int32_t code) {
-  SSyncQueryParam* pParam = param;
+static void syncQueryFn(void *param, void *res, int32_t code) {
+  SSyncQueryParam *pParam = param;
   pParam->pRequest = res;
   pParam->pRequest->code = code;
 
@@ -190,10 +190,10 @@ TAOS_RES *taos_query(TAOS *taos, const char *sql) {
     return NULL;
   }
 
-  STscObj* pTscObj = (STscObj*)taos;
+  STscObj *pTscObj = (STscObj *)taos;
 
 #if SYNC_ON_TOP_OF_ASYNC
-  SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
+  SSyncQueryParam *param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
   tsem_init(&param->sem, 0, 0);
 
   taos_query_a(pTscObj, sql, syncQueryFn, param);
@@ -606,16 +606,30 @@ const char *taos_get_server_info(TAOS *taos) {
 }
 
 typedef struct SqlParseWrapper {
-  SParseContext* pCtx;
+  SParseContext *pCtx;
   SCatalogReq    catalogReq;
-  SRequestObj*   pRequest;
-  SQuery*        pQuery;
+  SRequestObj   *pRequest;
+  SQuery        *pQuery;
 } SqlParseWrapper;
 
-void retrieveMetaCallback(SMetaData* pResultMeta, void* param, int32_t code) {
-  SqlParseWrapper *pWrapper = (SqlParseWrapper*) param;
-  SQuery* pQuery = pWrapper->pQuery;
-  SRequestObj* pRequest = pWrapper->pRequest;
+static void destorySqlParseWrapper(SqlParseWrapper *pWrapper) {
+  taosArrayDestroy(pWrapper->catalogReq.pDbVgroup);
+  taosArrayDestroy(pWrapper->catalogReq.pDbCfg);
+  taosArrayDestroy(pWrapper->catalogReq.pDbInfo);
+  taosArrayDestroy(pWrapper->catalogReq.pTableMeta);
+  taosArrayDestroy(pWrapper->catalogReq.pTableHash);
+  taosArrayDestroy(pWrapper->catalogReq.pUdf);
+  taosArrayDestroy(pWrapper->catalogReq.pIndex);
+  taosArrayDestroy(pWrapper->catalogReq.pUser);
+  taosArrayDestroy(pWrapper->catalogReq.pTableIndex);
+  taosMemoryFree(pWrapper->pCtx);
+  taosMemoryFree(pWrapper);
+}
+
+void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
+  SqlParseWrapper *pWrapper = (SqlParseWrapper *)param;
+  SQuery          *pQuery = pWrapper->pQuery;
+  SRequestObj     *pRequest = pWrapper->pRequest;
 
   if (code == TSDB_CODE_SUCCESS) {
     code = qAnalyseSqlSemantic(pWrapper->pCtx, &pWrapper->catalogReq, pResultMeta, pQuery);
@@ -630,22 +644,22 @@ void retrieveMetaCallback(SMetaData* pResultMeta, void* param, int32_t code) {
     TSWAP(pRequest->dbList, (pQuery)->pDbList);
     TSWAP(pRequest->tableList, (pQuery)->pTableList);
 
-    taosMemoryFree(pWrapper);
+    destorySqlParseWrapper(pWrapper);
     launchAsyncQuery(pRequest, pQuery);
   } else {
+    destorySqlParseWrapper(pWrapper);
     tscDebug("error happens, code:%d", code);
     if (NEED_CLIENT_HANDLE_ERROR(code)) {
-      tscDebug("0x%"PRIx64" client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code),
-               pRequest->retry, pRequest->requestId);
+      tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
+               pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
       pRequest->prevCode = code;
       doAsyncQuery(pRequest, true);
       return;
     }
 
     // return to app directly
-    taosMemoryFree(pWrapper);
-    tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self, tstrerror(code),
-             pRequest->requestId);
+    tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self,
+             tstrerror(code), pRequest->requestId);
     pRequest->code = code;
     pRequest->body.queryFp(pRequest->body.param, pRequest, code);
   }
@@ -682,7 +696,7 @@ void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param
   doAsyncQuery(pRequest, false);
 }
 
-int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
+int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
   const STscObj *pTscObj = pRequest->pTscObj;
 
   *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
@@ -690,7 +704,8 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  **pCxt = (SParseContext){.requestId = pRequest->requestId,
+  **pCxt = (SParseContext){
+      .requestId = pRequest->requestId,
       .requestRid = pRequest->self,
       .acctId = pTscObj->acctId,
       .db = pRequest->pDb,
@@ -704,12 +719,13 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
       .pUser = pTscObj->user,
       .schemalessType = pTscObj->schemalessType,
       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
-                           .async = true,};
+      .async = true,
+  };
   return TSDB_CODE_SUCCESS;
 }
 
-void doAsyncQuery(SRequestObj* pRequest, bool updateMetaForce) {
-  SParseContext* pCxt = NULL;
+void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
+  SParseContext *pCxt = NULL;
   STscObj       *pTscObj = pRequest->pTscObj;
   int32_t        code = 0;
 
@@ -753,23 +769,24 @@ void doAsyncQuery(SRequestObj* pRequest, bool updateMetaForce) {
                            .requestObjRefId = pCxt->requestRid,
                            .mgmtEps = pCxt->mgmtEpSet};
 
-  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, pRequest->requestId,
-                                &catalogReq, retrieveMetaCallback, pWrapper, &pRequest->body.queryJob);
+  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, pRequest->requestId, &catalogReq, retrieveMetaCallback, pWrapper,
+                                &pRequest->body.queryJob);
   if (code == TSDB_CODE_SUCCESS) {
     return;
   }
 
-  _error:
-  tscError("0x%"PRIx64" error happens, code:%d - %s, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code), pRequest->requestId);
+_error:
+  tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
+           pRequest->requestId);
   terrno = code;
   pRequest->code = code;
   pRequest->body.queryFp(pRequest->body.param, pRequest, code);
 }
 
-static void fetchCallback(void* pResult, void* param, int32_t code) {
-  SRequestObj* pRequest = (SRequestObj*) param;
+static void fetchCallback(void *pResult, void *param, int32_t code) {
+  SRequestObj *pRequest = (SRequestObj *)param;
 
-  SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
+  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
 
   pResultInfo->pData = pResult;
   pResultInfo->numOfRows = 0;
@@ -785,7 +802,8 @@ static void fetchCallback(void* pResult, void* param, int32_t code) {
     return;
   }
 
-  pRequest->code = setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp*)pResultInfo->pData, pResultInfo->convertUcs4, false);
+  pRequest->code =
+      setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, false);
   if (pRequest->code != TSDB_CODE_SUCCESS) {
     pResultInfo->numOfRows = 0;
     pRequest->code = code;
@@ -801,7 +819,7 @@ static void fetchCallback(void* pResult, void* param, int32_t code) {
 }
 
 void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
-  ASSERT (res != NULL && fp != NULL);
+  ASSERT(res != NULL && fp != NULL);
 
   SRequestObj *pRequest = res;
   pRequest->body.fetchFp = fp;
@@ -825,7 +843,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   schedulerAsyncFetchRows(pRequest->body.queryJob, fetchCallback, pRequest);
 }
 
-void taos_fetch_raw_block_a(TAOS_RES* res, __taos_async_fn_t fp, void* param) {
+void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   ASSERT(res != NULL && fp != NULL);
   SRequestObj *pRequest = res;
 
@@ -838,9 +856,9 @@ void taos_fetch_raw_block_a(TAOS_RES* res, __taos_async_fn_t fp, void* param) {
   taos_fetch_rows_a(res, fp, param);
 }
 
-const void* taos_get_raw_block(TAOS_RES* res) {
+const void *taos_get_raw_block(TAOS_RES *res) {
   ASSERT(res != NULL);
-  SRequestObj* pRequest = res;
+  SRequestObj *pRequest = res;
 
   return pRequest->body.resInfo.pData;
 }
@@ -924,10 +942,9 @@ int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
   return stmtSetTbTags(stmt, tags);
 }
 
-
 int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }
 
-int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
+int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
   if (stmt == NULL || NULL == fieldNum) {
     tscError("NULL parameter for %s", __FUNCTION__);
     terrno = TSDB_CODE_INVALID_PARA;
@@ -937,7 +954,7 @@ int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fiel
   return stmtGetTagFields(stmt, fieldNum, fields);
 }
 
-int taos_stmt_get_col_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
+int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
   if (stmt == NULL || NULL == fieldNum) {
     tscError("NULL parameter for %s", __FUNCTION__);
     terrno = TSDB_CODE_INVALID_PARA;
@@ -1102,4 +1119,3 @@ int taos_stmt_close(TAOS_STMT *stmt) {
 
   return stmtClose(stmt);
 }
-

source/common/src/tmsg.c

@@ -692,6 +692,7 @@ int32_t tDeserializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq
 void tFreeSMAltertbReq(SMAlterStbReq *pReq) {
   taosArrayDestroy(pReq->pFields);
   pReq->pFields = NULL;
+  taosMemoryFreeClear(pReq->comment);
 }
 
 int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) {
@@ -4064,6 +4065,7 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
   if (tEncodeI32(&encoder, sqlLen) < 0) return -1;
   if (tEncodeI32(&encoder, astLen) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->triggerType) < 0) return -1;
+  if (tEncodeI64(&encoder, pReq->maxDelay) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->watermark) < 0) return -1;
   if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
   if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
@@ -4090,6 +4092,7 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
   if (tDecodeI32(&decoder, &sqlLen) < 0) return -1;
   if (tDecodeI32(&decoder, &astLen) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->triggerType) < 0) return -1;
+  if (tDecodeI64(&decoder, &pReq->maxDelay) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->watermark) < 0) return -1;
 
   if (sqlLen > 0) {

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

@@ -77,7 +77,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     goto END;
   }
-  SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+  SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
 
   int32_t opNum = LIST_LENGTH(inner->pNodeList);
   if (opNum != 1) {
@@ -85,7 +85,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
     goto END;
   }
 
-  SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
+  SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
   if (qSubPlanToString(plan, pDst, pDstLen) < 0) {
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
     goto END;
@@ -93,7 +93,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
 
 END:
   if (pAst) nodesDestroyNode(pAst);
-  if (pPlan) nodesDestroyNode(pPlan);
+  if (pPlan) nodesDestroyNode((SNode*)pPlan);
   return terrno;
 }
 
@@ -378,8 +378,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
       SArray* taskInnerLevel = taosArrayInit(0, sizeof(void*));
       taosArrayPush(pStream->tasks, &taskInnerLevel);
 
-      SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
-      SSubplan*      plan = nodesListGetNode(inner->pNodeList, 0);
+      SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
+      SSubplan*      plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
       ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
 
       pFinalTask = tNewSStreamTask(pStream->uid);
@@ -407,8 +407,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
     SArray* taskSourceLevel = taosArrayInit(0, sizeof(void*));
     taosArrayPush(pStream->tasks, &taskSourceLevel);
 
-    SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 1);
-    SSubplan*      plan = nodesListGetNode(inner->pNodeList, 0);
+    SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 1);
+    SSubplan*      plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
     ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
 
     void* pIter = NULL;
@@ -449,9 +449,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
     SArray* taskOneLevel = taosArrayInit(0, sizeof(void*));
     taosArrayPush(pStream->tasks, &taskOneLevel);
 
-    SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+    SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
     ASSERT(LIST_LENGTH(inner->pNodeList) == 1);
-    SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
+    SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
     ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
 
     void* pIter = NULL;
@@ -509,7 +509,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
       return -1;
     }
 
-    SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+    SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
 
     int32_t opNum = LIST_LENGTH(inner->pNodeList);
     if (opNum != 1) {
@@ -517,7 +517,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
       terrno = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
       return -1;
     }
-    plan = nodesListGetNode(inner->pNodeList, 0);
+    plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
   }
 
   ASSERT(pSub->unassignedVgs);

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

@@ -235,10 +235,10 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
   }
 
   if (TSDB_CODE_SUCCESS == code) {
-    code = nodesNodeToString(pPlan, false, pStr, NULL);
+    code = nodesNodeToString((SNode*)pPlan, false, pStr, NULL);
   }
   nodesDestroyNode(pAst);
-  nodesDestroyNode(pPlan);
+  nodesDestroyNode((SNode*)pPlan);
   terrno = code;
   return code;
 }

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

@@ -387,7 +387,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       return -1;
     }
 
-    if (nodesNodeToString(pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
+    if (nodesNodeToString((SNode*)pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
       taosMemoryFree(topicObj.ast);
       taosMemoryFree(topicObj.sql);

source/libs/command/src/explain.c

@@ -211,7 +211,7 @@ int32_t qExplainGenerateResChildren(SPhysiNode *pNode, SExplainGroup *group, SNo
   SExplainResNode *pResNode = NULL;
   FOREACH(node, pPhysiChildren) {
     QRY_ERR_RET(qExplainGenerateResNode((SPhysiNode *)node, group, &pResNode));
-    QRY_ERR_RET(nodesListAppend(*pChildren, pResNode));
+    QRY_ERR_RET(nodesListAppend(*pChildren, (SNode*)pResNode));
   }
 
   return TSDB_CODE_SUCCESS;
@@ -715,7 +715,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         EXPLAIN_ROW_NEW(level + 1, "Sort Key: ");
         if (pResNode->pExecInfo) {
           for (int32_t i = 0; i < LIST_LENGTH(pSortNode->pSortKeys); ++i) {
-            SOrderByExprNode *ptn = nodesListGetNode(pSortNode->pSortKeys, i);
+            SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pSortNode->pSortKeys, i);
             EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
           }
         }
@@ -1039,7 +1039,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         EXPLAIN_ROW_NEW(level + 1, "Merge Key: ");
         if (pResNode->pExecInfo) {
           for (int32_t i = 0; i < LIST_LENGTH(pMergeNode->pMergeKeys); ++i) {
-            SOrderByExprNode *ptn = nodesListGetNode(pMergeNode->pMergeKeys, i);
+            SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pMergeNode->pMergeKeys, i);
             EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
           }
         }
@@ -1078,7 +1078,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
 
         EXPLAIN_ROW_NEW(level + 1, EXPLAIN_MERGE_KEYS_FORMAT);
         for (int32_t i = 0; i < LIST_LENGTH(pMergeNode->pMergeKeys); ++i) {
-          SOrderByExprNode *ptn = nodesListGetNode(pMergeNode->pMergeKeys, i);
+          SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pMergeNode->pMergeKeys, i);
           EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
         }
         EXPLAIN_ROW_END();

source/libs/executor/src/executorimpl.c

@@ -202,7 +202,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData idata = {{0}};
-    SSlotDescNode*  pDescNode = nodesListGetNode(pNode->pSlots, i);
+    SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
     //    if (!pDescNode->output) {  // todo disable it temporarily
     //      continue;
     //    }
@@ -2920,7 +2920,7 @@ static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo*
   }
 
   for (int32_t i = 0; i < numOfSources; ++i) {
-    SNodeListNode* pNode = nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
+    SNodeListNode* pNode = (SNodeListNode*)nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
     taosArrayPush(pInfo->pSources, pNode);
   }
 
@@ -4494,7 +4494,7 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
         if (NULL == res) {  // todo handle error
         } else {
           res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
-          nodesListAppend(pFuncNode->pParameterList, res);
+          nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
         }
       }
 #endif

source/libs/executor/src/timewindowoperator.c

@@ -1378,7 +1378,7 @@ void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
       taosMemoryFreeClear(pChildOp);
     }
   }
-  nodesDestroyNode(pInfo->pPhyNode);
+  nodesDestroyNode((SNode*)pInfo->pPhyNode);
 }
 
 static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
@@ -2157,7 +2157,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
   pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
   pInfo->pUpdateRes->info.type = STREAM_REPROCESS;
   blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
-  pInfo->pPhyNode = nodesCloneNode(pPhyNode);
+  pInfo->pPhyNode = (SPhysiNode*)nodesCloneNode((SNode*)pPhyNode);
 
   pOperator->name = "StreamFinalIntervalOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;

source/libs/function/inc/functionMgtInt.h

@@ -42,6 +42,7 @@ extern "C" {
 #define FUNC_MGT_SELECT_FUNC            FUNC_MGT_FUNC_CLASSIFICATION_MASK(13)
 #define FUNC_MGT_REPEAT_SCAN_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(14)
 #define FUNC_MGT_FORBID_FILL_FUNC       FUNC_MGT_FUNC_CLASSIFICATION_MASK(15)
+#define FUNC_MGT_FORBID_STREAM_FUNC     FUNC_MGT_FUNC_CLASSIFICATION_MASK(16)
 
 #define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 

source/libs/function/src/builtins.c

@@ -1274,7 +1274,7 @@ void static addTimezoneParam(SNodeList* pList) {
   varDataSetLen(pVal->datum.p, len);
   strncpy(varDataVal(pVal->datum.p), pVal->literal, len);
 
-  nodesListAppend(pList, pVal);
+  nodesListAppend(pList, (SNode*)pVal);
 }
 
 static int32_t translateToIso8601(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
@@ -1363,7 +1363,7 @@ static int32_t translateToJson(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  SExprNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
+  SExprNode* pPara = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
   if (QUERY_NODE_VALUE != nodeType(pPara) || (!IS_VAR_DATA_TYPE(pPara->resType.type))) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }

source/libs/function/src/functionMgt.c

@@ -161,6 +161,8 @@ bool fmIsUserDefinedFunc(int32_t funcId) { return funcId > FUNC_UDF_ID_START; }
 
 bool fmIsForbidFillFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_FILL_FUNC); }
 
+bool fmIsForbidStreamFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_STREAM_FUNC); }
+
 void fmFuncMgtDestroy() {
   void* m = gFunMgtService.pFuncNameHashTable;
   if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
@@ -239,7 +241,7 @@ static int32_t getFuncInfo(SFunctionNode* pFunc) {
 }
 
 static SFunctionNode* createFunction(const char* pName, SNodeList* pParameterList) {
-  SFunctionNode* pFunc = nodesMakeNode(QUERY_NODE_FUNCTION);
+  SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
   if (NULL == pFunc) {
     return NULL;
   }
@@ -247,14 +249,14 @@ static SFunctionNode* createFunction(const char* pName, SNodeList* pParameterLis
   pFunc->pParameterList = pParameterList;
   if (TSDB_CODE_SUCCESS != getFuncInfo(pFunc)) {
     pFunc->pParameterList = NULL;
-    nodesDestroyNode(pFunc);
+    nodesDestroyNode((SNode*)pFunc);
     return NULL;
   }
   return pFunc;
 }
 
 static SColumnNode* createColumnByFunc(const SFunctionNode* pFunc) {
-  SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
+  SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
   if (NULL == pCol) {
     return NULL;
   }
@@ -291,7 +293,7 @@ static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNod
 static int32_t createMergeFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
                                    SFunctionNode** pMergeFunc) {
   SNodeList* pParameterList = NULL;
-  nodesListMakeStrictAppend(&pParameterList, createColumnByFunc(pPartialFunc));
+  nodesListMakeStrictAppend(&pParameterList, (SNode*)createColumnByFunc(pPartialFunc));
   *pMergeFunc = createFunction(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pParameterList);
   if (NULL == *pMergeFunc) {
     nodesDestroyList(pParameterList);
@@ -316,8 +318,8 @@ int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc
   }
 
   if (TSDB_CODE_SUCCESS != code) {
-    nodesDestroyNode(*pPartialFunc);
-    nodesDestroyNode(*pMergeFunc);
+    nodesDestroyNode((SNode*)*pPartialFunc);
+    nodesDestroyNode((SNode*)*pMergeFunc);
   }
 
   return code;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -40,6 +40,10 @@
       break;                                   \
     }                                          \
     (pDst)->fldname = strdup((pSrc)->fldname); \
+    if (NULL == (pDst)->fldname) {             \
+      nodesDestroyNode((SNode*)pDst);          \
+      return NULL;                             \
+    }                                          \
   } while (0)
 
 #define CLONE_NODE_FIELD(fldname)                      \
@@ -49,7 +53,19 @@
     }                                                  \
     (pDst)->fldname = nodesCloneNode((pSrc)->fldname); \
     if (NULL == (pDst)->fldname) {                     \
-      nodesDestroyNode((SNode*)(pDst));                \
+      nodesDestroyNode((SNode*)pDst);                  \
+      return NULL;                                     \
+    }                                                  \
+  } while (0)
+
+#define CLONE_NODE_FIELD_EX(fldname, nodePtrType)                           \
+  do {                                                                      \
+    if (NULL == (pSrc)->fldname) {                                          \
+      break;                                                                \
+    }                                                                       \
+    (pDst)->fldname = (nodePtrType)nodesCloneNode((SNode*)(pSrc)->fldname); \
+    if (NULL == (pDst)->fldname) {                                          \
+      nodesDestroyNode((SNode*)pDst);                                       \
       return NULL;                                                          \
     }                                                                       \
   } while (0)
@@ -61,7 +77,7 @@
     }                                                  \
     (pDst)->fldname = nodesCloneList((pSrc)->fldname); \
     if (NULL == (pDst)->fldname) {                     \
-      nodesDestroyNode((SNode*)(pDst));                \
+      nodesDestroyNode((SNode*)pDst);                  \
       return NULL;                                     \
     }                                                  \
   } while (0)
@@ -73,7 +89,7 @@
     }                                             \
     (pDst)->fldname = cloneFunc((pSrc)->fldname); \
     if (NULL == (pDst)->fldname) {                \
-      nodesDestroyNode((SNode*)(pDst));           \
+      nodesDestroyNode((SNode*)pDst);             \
       return NULL;                                \
     }                                             \
   } while (0)
@@ -81,6 +97,7 @@
 #define COPY_BASE_OBJECT_FIELD(fldname, copyFunc)                   \
   do {                                                              \
     if (NULL == copyFunc(&((pSrc)->fldname), &((pDst)->fldname))) { \
+      nodesDestroyNode((SNode*)pDst);                               \
       return NULL;                                                  \
     }                                                               \
   } while (0)
@@ -147,7 +164,7 @@ static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
       int32_t len = varDataTLen(pSrc->datum.p) + 1;
       pDst->datum.p = taosMemoryCalloc(1, len);
       if (NULL == pDst->datum.p) {
-        nodesDestroyNode(pDst);
+        nodesDestroyNode((SNode*)pDst);
         return NULL;
       }
       memcpy(pDst->datum.p, pSrc->datum.p, len);
@@ -275,8 +292,8 @@ static SNode* stateWindowNodeCopy(const SStateWindowNode* pSrc, SStateWindowNode
 }
 
 static SNode* sessionWindowNodeCopy(const SSessionWindowNode* pSrc, SSessionWindowNode* pDst) {
-  CLONE_NODE_FIELD(pCol);
-  CLONE_NODE_FIELD(pGap);
+  CLONE_NODE_FIELD_EX(pCol, SColumnNode*);
+  CLONE_NODE_FIELD_EX(pGap, SValueNode*);
   return (SNode*)pDst;
 }
 
@@ -442,7 +459,7 @@ static SNode* logicIndefRowsFuncCopy(const SIndefRowsFuncLogicNode* pSrc, SIndef
 
 static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
   COPY_OBJECT_FIELD(id, sizeof(SSubplanId));
-  CLONE_NODE_FIELD(pNode);
+  CLONE_NODE_FIELD_EX(pNode, SLogicNode*);
   COPY_SCALAR_FIELD(subplanType);
   COPY_SCALAR_FIELD(level);
   COPY_SCALAR_FIELD(splitFlag);
@@ -450,7 +467,7 @@ static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
 }
 
 static SNode* physiNodeCopy(const SPhysiNode* pSrc, SPhysiNode* pDst) {
-  CLONE_NODE_FIELD(pOutputDataBlockDesc);
+  CLONE_NODE_FIELD_EX(pOutputDataBlockDesc, SDataBlockDescNode*);
   CLONE_NODE_FIELD(pConditions);
   CLONE_NODE_LIST_FIELD(pChildren);
   return (SNode*)pDst;
@@ -555,8 +572,8 @@ static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
   CLONE_NODE_LIST_FIELD(pGroupByList);
   CLONE_NODE_FIELD(pHaving);
   CLONE_NODE_LIST_FIELD(pOrderByList);
-  CLONE_NODE_FIELD(pLimit);
-  CLONE_NODE_FIELD(pLimit);
+  CLONE_NODE_FIELD_EX(pLimit, SLimitNode*);
+  CLONE_NODE_FIELD_EX(pLimit, SLimitNode*);
   COPY_CHAR_ARRAY_FIELD(stmtName);
   COPY_SCALAR_FIELD(precision);
   COPY_SCALAR_FIELD(isEmptyResult);
@@ -566,7 +583,7 @@ static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
   return (SNode*)pDst;
 }
 
-SNodeptr nodesCloneNode(const SNodeptr pNode) {
+SNode* nodesCloneNode(const SNode* pNode) {
   if (NULL == pNode) {
     return NULL;
   }

source/libs/nodes/src/nodesCodeFuncs.c

@@ -4205,7 +4205,7 @@ static int32_t jsonToNodeObject(const SJson* pJson, const char* pName, SNode** p
   return makeNodeByJson(pJsonNode, pNode);
 }
 
-int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen) {
+int32_t nodesNodeToString(const SNode* pNode, bool format, char** pStr, int32_t* pLen) {
   if (NULL == pNode || NULL == pStr) {
     terrno = TSDB_CODE_FAILED;
     return TSDB_CODE_FAILED;

source/libs/nodes/src/nodesEqualFuncs.c

@@ -137,7 +137,7 @@ static bool functionNodeEqual(const SFunctionNode* a, const SFunctionNode* b) {
   return true;
 }
 
-bool nodesEqualNode(const SNodeptr a, const SNodeptr b) {
+bool nodesEqualNode(const SNode* a, const SNode* b) {
   if (a == b) {
     return true;
   }

source/libs/nodes/src/nodesTraverseFuncs.c

@@ -168,7 +168,7 @@ static EDealRes walkExprs(SNodeList* pNodeList, ETraversalOrder order, FNodeWalk
   return DEAL_RES_CONTINUE;
 }
 
-void nodesWalkExpr(SNodeptr pNode, FNodeWalker walker, void* pContext) {
+void nodesWalkExpr(SNode* pNode, FNodeWalker walker, void* pContext) {
   (void)walkExpr(pNode, TRAVERSAL_PREORDER, walker, pContext);
 }
 
@@ -176,7 +176,7 @@ void nodesWalkExprs(SNodeList* pNodeList, FNodeWalker walker, void* pContext) {
   (void)walkExprs(pNodeList, TRAVERSAL_PREORDER, walker, pContext);
 }
 
-void nodesWalkExprPostOrder(SNodeptr pNode, FNodeWalker walker, void* pContext) {
+void nodesWalkExprPostOrder(SNode* pNode, FNodeWalker walker, void* pContext) {
   (void)walkExpr(pNode, TRAVERSAL_POSTORDER, walker, pContext);
 }
 

source/libs/parser/inc/parInt.h

@@ -27,12 +27,12 @@ extern "C" {
 #define QUERY_SMA_OPTIMIZE_DISABLE 0
 #define QUERY_SMA_OPTIMIZE_ENABLE  1
 
-int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery);
-int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery);
+int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache);
+int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache);
 int32_t parse(SParseContext* pParseCxt, SQuery** pQuery);
-int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery);
-int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery);
-int32_t translate(SParseContext* pParseCxt, SQuery* pQuery);
+int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
+int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
+int32_t translate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
 int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
 int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery);
 

source/libs/parser/inc/parUtil.h

@@ -87,6 +87,7 @@ int32_t getUserAuthFromCache(SParseMetaCache* pMetaCache, const char* pUser, con
                              bool* pPass);
 int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFuncInfo* pInfo);
 int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes);
+void    destoryParseMetaCache(SParseMetaCache* pMetaCache);
 
 #ifdef __cplusplus
 }

source/libs/parser/inc/sql.y

@@ -461,6 +461,7 @@ into_opt(A) ::= INTO full_table_name(B).
 stream_options(A) ::= .                                                           { A = createStreamOptions(pCxt); }
 stream_options(A) ::= stream_options(B) TRIGGER AT_ONCE.                          { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_AT_ONCE; A = B; }
 stream_options(A) ::= stream_options(B) TRIGGER WINDOW_CLOSE.                     { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_WINDOW_CLOSE; A = B; }
+stream_options(A) ::= stream_options(B) TRIGGER MAX_DELAY duration_literal(C).    { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_MAX_DELAY; ((SStreamOptions*)B)->pDelay = releaseRawExprNode(pCxt, C); A = B; }
 stream_options(A) ::= stream_options(B) WATERMARK duration_literal(C).            { ((SStreamOptions*)B)->pWatermark = releaseRawExprNode(pCxt, C); A = B; }
 
 /************************************************ kill connection/query ***********************************************/

source/libs/parser/src/parAuthenticator.c

@@ -105,7 +105,7 @@ static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery) {
-  SAuthCxt cxt = {.pParseCxt = pParseCxt, .pMetaCache = pQuery->pMetaCache, .errCode = TSDB_CODE_SUCCESS};
+int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache) {
+  SAuthCxt cxt = {.pParseCxt = pParseCxt, .pMetaCache = pMetaCache, .errCode = TSDB_CODE_SUCCESS};
   return authQuery(&cxt, pQuery->pRoot);
 }