Merge pull request #19887 from taosdata/feat/ly_ig_udpate

feat(stream):add ignore check update option for create stream

include/common/tmsg.h

@@ -1754,6 +1754,7 @@ typedef struct {
 #define STREAM_FILL_HISTORY_ON        1
 #define STREAM_FILL_HISTORY_OFF       0
 #define STREAM_DEFAULT_FILL_HISTORY   STREAM_FILL_HISTORY_OFF
+#define STREAM_DEFAULT_IGNORE_UPDATE  0
 
 typedef struct {
   char    name[TSDB_STREAM_FNAME_LEN];
@@ -1771,6 +1772,7 @@ typedef struct {
   SArray* pTags;  // array of SField
   // 3.0.20
   int64_t checkpointFreq;  // ms
+  int8_t  igUpdate;
 } SCMCreateStreamReq;
 
 typedef struct {

include/common/ttokendef.h

@@ -209,138 +209,137 @@
 #define TK_IGNORE               191
 #define TK_EXPIRED              192
 #define TK_FILL_HISTORY         193
-#define TK_SUBTABLE                       194
-#define TK_KILL                           195
-#define TK_CONNECTION                     196
-#define TK_TRANSACTION                    197
-#define TK_BALANCE                        198
-#define TK_VGROUP                         199
-#define TK_MERGE                          200
-#define TK_REDISTRIBUTE                   201
-#define TK_SPLIT                          202
-#define TK_DELETE                         203
-#define TK_INSERT                         204
-#define TK_NULL                           205
-#define TK_NK_QUESTION                    206
-#define TK_NK_ARROW                       207
-#define TK_ROWTS                          208
-#define TK_QSTART                         209
-#define TK_QEND                           210
-#define TK_QDURATION                      211
-#define TK_WSTART                         212
-#define TK_WEND                           213
-#define TK_WDURATION                      214
-#define TK_IROWTS                         215
-#define TK_CAST                           216
-#define TK_NOW                            217
-#define TK_TODAY                          218
-#define TK_TIMEZONE                       219
-#define TK_CLIENT_VERSION                 220
-#define TK_SERVER_VERSION                 221
-#define TK_SERVER_STATUS                  222
-#define TK_CURRENT_USER                   223
-#define TK_COUNT                          224
-#define TK_LAST_ROW                       225
-#define TK_CASE                           226
-#define TK_END                            227
-#define TK_WHEN                           228
-#define TK_THEN                           229
-#define TK_ELSE                           230
-#define TK_BETWEEN                        231
-#define TK_IS                             232
-#define TK_NK_LT                          233
-#define TK_NK_GT                          234
-#define TK_NK_LE                          235
-#define TK_NK_GE                          236
-#define TK_NK_NE                          237
-#define TK_MATCH                          238
-#define TK_NMATCH                         239
-#define TK_CONTAINS                       240
-#define TK_IN                             241
-#define TK_JOIN                           242
-#define TK_INNER                          243
-#define TK_SELECT                         244
-#define TK_DISTINCT                       245
-#define TK_WHERE                          246
-#define TK_PARTITION                      247
-#define TK_BY                             248
-#define TK_SESSION                        249
-#define TK_STATE_WINDOW                   250
-#define TK_SLIDING                        251
-#define TK_FILL                           252
-#define TK_VALUE                          253
-#define TK_VALUE_F                        254
-#define TK_NONE                           255
-#define TK_PREV                           256
-#define TK_NULL_F                         257
-#define TK_LINEAR                         258
-#define TK_NEXT                           259
-#define TK_HAVING                         260
-#define TK_RANGE                          261
-#define TK_EVERY                          262
-#define TK_ORDER                          263
-#define TK_SLIMIT                         264
-#define TK_SOFFSET                        265
-#define TK_LIMIT                          266
-#define TK_OFFSET                         267
-#define TK_ASC                            268
-#define TK_NULLS                          269
-#define TK_ABORT                          270
-#define TK_AFTER                          271
-#define TK_ATTACH                         272
-#define TK_BEFORE                         273
-#define TK_BEGIN                          274
-#define TK_BITAND                         275
-#define TK_BITNOT                         276
-#define TK_BITOR                          277
-#define TK_BLOCKS                         278
-#define TK_CHANGE                         279
-#define TK_COMMA                          280
-#define TK_COMPACT                        281
-#define TK_CONCAT                         282
-#define TK_CONFLICT                       283
-#define TK_COPY                           284
-#define TK_DEFERRED                       285
-#define TK_DELIMITERS                     286
-#define TK_DETACH                         287
-#define TK_DIVIDE                         288
-#define TK_DOT                            289
-#define TK_EACH                           290
-#define TK_FAIL                           291
-#define TK_FILE                           292
-#define TK_FOR                            293
-#define TK_GLOB                           294
-#define TK_ID                             295
-#define TK_IMMEDIATE                      296
-#define TK_IMPORT                         297
-#define TK_INITIALLY                      298
-#define TK_INSTEAD                        299
-#define TK_ISNULL                         300
-#define TK_KEY                            301
-#define TK_MODULES                        302
-#define TK_NK_BITNOT                      303
-#define TK_NK_SEMI                        304
-#define TK_NOTNULL                        305
-#define TK_OF                             306
-#define TK_PLUS                           307
-#define TK_PRIVILEGE                      308
-#define TK_RAISE                          309
-#define TK_REPLACE                        310
-#define TK_RESTRICT                       311
-#define TK_ROW                            312
-#define TK_SEMI                           313
-#define TK_STAR                           314
-#define TK_STATEMENT                      315
-#define TK_STRICT                         316
-#define TK_STRING                         317
-#define TK_TIMES                          318
-#define TK_UPDATE                         319
+#define TK_UPDATE               194
+#define TK_SUBTABLE             195
+#define TK_KILL                 196
+#define TK_CONNECTION           197
+#define TK_TRANSACTION          198
+#define TK_BALANCE              199
+#define TK_VGROUP               200
+#define TK_MERGE                201
+#define TK_REDISTRIBUTE         202
+#define TK_SPLIT                203
+#define TK_DELETE               204
+#define TK_INSERT               205
+#define TK_NULL                 206
+#define TK_NK_QUESTION          207
+#define TK_NK_ARROW             208
+#define TK_ROWTS                209
+#define TK_QSTART               210
+#define TK_QEND                 211
+#define TK_QDURATION            212
+#define TK_WSTART               213
+#define TK_WEND                 214
+#define TK_WDURATION            215
+#define TK_IROWTS               216
+#define TK_CAST                 217
+#define TK_NOW                  218
+#define TK_TODAY                219
+#define TK_TIMEZONE             220
+#define TK_CLIENT_VERSION       221
+#define TK_SERVER_VERSION       222
+#define TK_SERVER_STATUS        223
+#define TK_CURRENT_USER         224
+#define TK_COUNT                225
+#define TK_LAST_ROW             226
+#define TK_CASE                 227
+#define TK_END                  228
+#define TK_WHEN                 229
+#define TK_THEN                 230
+#define TK_ELSE                 231
+#define TK_BETWEEN              232
+#define TK_IS                   233
+#define TK_NK_LT                234
+#define TK_NK_GT                235
+#define TK_NK_LE                236
+#define TK_NK_GE                237
+#define TK_NK_NE                238
+#define TK_MATCH                239
+#define TK_NMATCH               240
+#define TK_CONTAINS             241
+#define TK_IN                   242
+#define TK_JOIN                 243
+#define TK_INNER                244
+#define TK_SELECT               245
+#define TK_DISTINCT             246
+#define TK_WHERE                247
+#define TK_PARTITION            248
+#define TK_BY                   249
+#define TK_SESSION              250
+#define TK_STATE_WINDOW         251
+#define TK_SLIDING              252
+#define TK_FILL                 253
+#define TK_VALUE                254
+#define TK_VALUE_F              255
+#define TK_NONE                 256
+#define TK_PREV                 257
+#define TK_NULL_F               258
+#define TK_LINEAR               259
+#define TK_NEXT                 260
+#define TK_HAVING               261
+#define TK_RANGE                262
+#define TK_EVERY                263
+#define TK_ORDER                264
+#define TK_SLIMIT               265
+#define TK_SOFFSET              266
+#define TK_LIMIT                267
+#define TK_OFFSET               268
+#define TK_ASC                  269
+#define TK_NULLS                270
+#define TK_ABORT                271
+#define TK_AFTER                272
+#define TK_ATTACH               273
+#define TK_BEFORE               274
+#define TK_BEGIN                275
+#define TK_BITAND               276
+#define TK_BITNOT               277
+#define TK_BITOR                278
+#define TK_BLOCKS               279
+#define TK_CHANGE               280
+#define TK_COMMA                281
+#define TK_COMPACT              282
+#define TK_CONCAT               283
+#define TK_CONFLICT             284
+#define TK_COPY                 285
+#define TK_DEFERRED             286
+#define TK_DELIMITERS           287
+#define TK_DETACH               288
+#define TK_DIVIDE               289
+#define TK_DOT                  290
+#define TK_EACH                 291
+#define TK_FAIL                 292
+#define TK_FILE                 293
+#define TK_FOR                  294
+#define TK_GLOB                 295
+#define TK_ID                   296
+#define TK_IMMEDIATE            297
+#define TK_IMPORT               298
+#define TK_INITIALLY            299
+#define TK_INSTEAD              300
+#define TK_ISNULL               301
+#define TK_KEY                  302
+#define TK_MODULES              303
+#define TK_NK_BITNOT            304
+#define TK_NK_SEMI              305
+#define TK_NOTNULL              306
+#define TK_OF                   307
+#define TK_PLUS                 308
+#define TK_PRIVILEGE            309
+#define TK_RAISE                310
+#define TK_REPLACE              311
+#define TK_RESTRICT             312
+#define TK_ROW                  313
+#define TK_SEMI                 314
+#define TK_STAR                 315
+#define TK_STATEMENT            316
+#define TK_STRICT               317
+#define TK_STRING               318
+#define TK_TIMES                319
 #define TK_VALUES               320
 #define TK_VARIABLE             321
 #define TK_VIEW                 322
 #define TK_WAL                  323
 
-
 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601
 #define TK_NK_ILLEGAL 602

include/libs/nodes/cmdnodes.h

@@ -389,6 +389,7 @@ typedef struct SStreamOptions {
   SNode*    pDeleteMark;
   int8_t    fillHistory;
   int8_t    ignoreExpired;
+  int8_t    ignoreUpdate;
 } SStreamOptions;
 
 typedef struct SCreateStreamStmt {

include/libs/nodes/plannodes.h

@@ -93,6 +93,7 @@ typedef struct SScanLogicNode {
   int64_t       watermark;
   int64_t       deleteMark;
   int8_t        igExpired;
+  int8_t        igCheckUpdate;
   SArray*       pSmaIndexes;
   SNodeList*    pGroupTags;
   bool          groupSort;
@@ -217,6 +218,7 @@ typedef struct SWindowLogicNode {
   int64_t          watermark;
   int64_t          deleteMark;
   int8_t           igExpired;
+  int8_t           igCheckUpdate;
   EWindowAlgorithm windowAlgo;
   EOrder           inputTsOrder;
   EOrder           outputTsOrder;
@@ -357,6 +359,7 @@ typedef struct STableScanPhysiNode {
   int64_t        watermark;
   int8_t         igExpired;
   bool           assignBlockUid;
+  int8_t         igCheckUpdate;
 } STableScanPhysiNode;
 
 typedef STableScanPhysiNode STableSeqScanPhysiNode;

include/libs/planner/planner.h

@@ -36,6 +36,7 @@ typedef struct SPlanContext {
   int64_t     watermark;
   int64_t     deleteMark;
   int8_t      igExpired;
+  int8_t      igCheckUpdate;
   char*       pMsg;
   int32_t     msgLen;
   const char* pUser;

source/common/src/tmsg.c

@@ -5425,6 +5425,7 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
     if (tEncodeI32(&encoder, pField->bytes) < 0) return -1;
     if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
   }
+  if (tEncodeI8(&encoder, pReq->igUpdate) < 0) return -1;
 
   tEndEncode(&encoder);
 
@@ -5486,6 +5487,8 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
     }
   }
 
+  if (tDecodeI8(&decoder, &pReq->igUpdate) < 0) return -1;
+
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);

source/dnode/mnode/impl/inc/mndDef.h

@@ -648,6 +648,7 @@ typedef struct {
   int64_t checkpointFreq;  // ms
   int64_t currentTick;     // do not serialize
   int64_t deleteMark;
+  int8_t  igCheckUpdate;
 } SStreamObj;
 
 int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);

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

@@ -78,6 +78,7 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
 
   // 3.0.20
   if (tEncodeI64(pEncoder, pObj->checkpointFreq) < 0) return -1;
+  if (tEncodeI8(pEncoder, pObj->igCheckUpdate) < 0) return -1;
 
   tEndEncode(pEncoder);
   return pEncoder->pos;
@@ -145,6 +146,7 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {
   // 3.0.20
   if (sver >= 2) {
     if (tDecodeI64(pDecoder, &pObj->checkpointFreq) < 0) return -1;
+    if (tDecodeI8(pDecoder, &pObj->igCheckUpdate) < 0) return -1;
   }
   tEndDecode(pDecoder);
   return 0;
@@ -489,7 +491,7 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
     tlen += tEncodeSMqConsumerEp(buf, pConsumerEp);
     cnt++;
   }
-  if(cnt != sz) return -1;
+  if (cnt != sz) return -1;
   tlen += taosEncodeArray(buf, pSub->unassignedVgs, (FEncode)tEncodeSMqVgEp);
   tlen += taosEncodeString(buf, pSub->dbName);
   return tlen;

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

@@ -297,6 +297,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
   pObj->triggerParam = pCreate->maxDelay;
   pObj->watermark = pCreate->watermark;
   pObj->fillHistory = pCreate->fillHistory;
+  pObj->igCheckUpdate = pCreate->igUpdate;
 
   memcpy(pObj->sourceDb, pCreate->sourceDB, TSDB_DB_FNAME_LEN);
   SDbObj *pSourceDb = mndAcquireDb(pMnode, pCreate->sourceDB);
@@ -345,6 +346,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
       .triggerType = pObj->trigger == STREAM_TRIGGER_MAX_DELAY ? STREAM_TRIGGER_WINDOW_CLOSE : pObj->trigger,
       .watermark = pObj->watermark,
       .igExpired = pObj->igExpired,
+      .igCheckUpdate = pObj->igCheckUpdate,
   };
 
   // using ast and param to build physical plan

source/libs/executor/inc/executorimpl.h

@@ -474,6 +474,8 @@ typedef struct SStreamScanInfo {
   int32_t blockRecoverContiCnt;
   int32_t blockRecoverTotCnt;
 
+  int8_t igCheckUpdate;
+  int8_t igExpired;
 } SStreamScanInfo;
 
 typedef struct {

source/libs/executor/src/groupoperator.c

@@ -1109,7 +1109,7 @@ void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->partitionSup = *pParSup;
   pScanInfo->pPartScalarSup = pExpr;
-  if (!pScanInfo->pUpdateInfo) {
+  if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
     pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, 0);
   }
 }

source/libs/executor/src/scanoperator.c

@@ -1435,7 +1435,12 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
     dumyInfo.cur.pageId = -1;
     bool        isClosed = false;
     STimeWindow win = {.skey = INT64_MIN, .ekey = INT64_MAX};
-    if (tableInserted && isOverdue(tsCol[rowId], &pInfo->twAggSup)) {
+    bool overDue = isOverdue(tsCol[rowId], &pInfo->twAggSup);
+    if (pInfo->igExpired && overDue) {
+      continue;
+    }
+
+    if (tableInserted && overDue) {
       win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);
       isClosed = isCloseWindow(&win, &pInfo->twAggSup);
     }
@@ -1701,41 +1706,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
   SStreamScanInfo* pInfo = pOperator->info;
 
   qDebug("stream scan called");
-#if 0
-  SStreamState* pState = pTaskInfo->streamInfo.pState;
-  if (pState) {
-    printf(">>>>>>>> stream write backend\n");
-    SWinKey key = {
-        .ts = 1,
-        .groupId = 2,
-    };
-    char tmp[100] = "abcdefg1";
-    if (streamStatePut(pState, &key, &tmp, strlen(tmp) + 1) < 0) {
-      ASSERT(0);
-    }
-
-    key.ts = 2;
-    char tmp2[100] = "abcdefg2";
-    if (streamStatePut(pState, &key, &tmp2, strlen(tmp2) + 1) < 0) {
-      ASSERT(0);
-    }
-
-    key.groupId = 5;
-    key.ts = 1;
-    char tmp3[100] = "abcdefg3";
-    if (streamStatePut(pState, &key, &tmp3, strlen(tmp3) + 1) < 0) {
-      ASSERT(0);
-    }
-
-    char*   val2 = NULL;
-    int32_t sz;
-    if (streamStateGet(pState, &key, (void**)&val2, &sz) < 0) {
-      ASSERT(0);
-    }
-    printf("stream read %s %d\n", val2, sz);
-    streamFreeVal(val2);
-  }
-#endif
 
   if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE1 ||
       pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE2) {
@@ -2368,6 +2338,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   pInfo->pUpdateDataRes = createSpecialDataBlock(STREAM_CLEAR);
   pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
   pInfo->partitionSup.needCalc = false;
+  pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;
+  pInfo->igExpired = pTableScanNode->igExpired;
+  pInfo->twAggSup.maxTs = INT64_MIN;
 
   setOperatorInfo(pOperator, "StreamScanOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,
                   pTaskInfo);

source/libs/executor/src/timewindowoperator.c

@@ -1726,7 +1726,7 @@ void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSuppor
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->windowSup.parentType = type;
   pScanInfo->windowSup.pIntervalAggSup = pSup;
-  if (!pScanInfo->pUpdateInfo) {
+  if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
     pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
   }
   pScanInfo->interval = *pInterval;
@@ -2893,7 +2893,7 @@ void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uin
   }
   SStreamScanInfo* pScanInfo = downstream->info;
   pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
-  if (!pScanInfo->pUpdateInfo) {
+  if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
     pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark);
   }
   pScanInfo->twAggSup = *pTwSup;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -380,6 +380,7 @@ static int32_t logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) {
   COPY_SCALAR_FIELD(watermark);
   COPY_SCALAR_FIELD(deleteMark);
   COPY_SCALAR_FIELD(igExpired);
+  COPY_SCALAR_FIELD(igCheckUpdate);
   CLONE_NODE_LIST_FIELD(pGroupTags);
   COPY_SCALAR_FIELD(groupSort);
   CLONE_NODE_LIST_FIELD(pTags);
@@ -467,6 +468,7 @@ static int32_t logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* p
   COPY_SCALAR_FIELD(watermark);
   COPY_SCALAR_FIELD(deleteMark);
   COPY_SCALAR_FIELD(igExpired);
+  COPY_SCALAR_FIELD(igCheckUpdate);
   COPY_SCALAR_FIELD(windowAlgo);
   COPY_SCALAR_FIELD(inputTsOrder);
   COPY_SCALAR_FIELD(outputTsOrder);

source/libs/nodes/src/nodesCodeFuncs.c

@@ -1553,6 +1553,7 @@ static const char* jkTableScanPhysiPlanGroupSort = "GroupSort";
 static const char* jkTableScanPhysiPlanTags = "Tags";
 static const char* jkTableScanPhysiPlanSubtable = "Subtable";
 static const char* jkTableScanPhysiPlanAssignBlockUid = "AssignBlockUid";
+static const char* jkTableScanPhysiPlanIgnoreUpdate = "IgnoreUpdate";
 
 static int32_t physiTableScanNodeToJson(const void* pObj, SJson* pJson) {
   const STableScanPhysiNode* pNode = (const STableScanPhysiNode*)pObj;
@@ -1618,6 +1619,9 @@ static int32_t physiTableScanNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddBoolToObject(pJson, jkTableScanPhysiPlanAssignBlockUid, pNode->assignBlockUid);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkTableScanPhysiPlanIgnoreUpdate, pNode->igCheckUpdate);
+  }
 
   return code;
 }
@@ -1686,6 +1690,9 @@ static int32_t jsonToPhysiTableScanNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBoolValue(pJson, jkTableScanPhysiPlanAssignBlockUid, &pNode->assignBlockUid);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetTinyIntValue(pJson, jkTableScanPhysiPlanIgnoreUpdate, &pNode->igCheckUpdate);
+  }
 
   return code;
 }

source/libs/nodes/src/nodesMsgFuncs.c

@@ -2078,6 +2078,9 @@ static int32_t physiTableScanNodeInlineToMsg(const void* pObj, STlvEncoder* pEnc
   if (TSDB_CODE_SUCCESS == code) {
     code = tlvEncodeValueBool(pEncoder, pNode->assignBlockUid);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeValueI8(pEncoder, pNode->igCheckUpdate);
+  }
 
   return code;
 }
@@ -2154,6 +2157,9 @@ static int32_t msgToPhysiTableScanNodeInline(STlvDecoder* pDecoder, void* pObj)
   if (TSDB_CODE_SUCCESS == code) {
     code = tlvDecodeValueBool(pDecoder, &pNode->assignBlockUid);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvDecodeValueI8(pDecoder, &pNode->igCheckUpdate);
+  }
 
   return code;
 }

source/libs/parser/inc/sql.y

@@ -544,6 +544,7 @@ stream_options(A) ::= stream_options(B) TRIGGER MAX_DELAY duration_literal(C).
 stream_options(A) ::= stream_options(B) WATERMARK duration_literal(C).            { ((SStreamOptions*)B)->pWatermark = releaseRawExprNode(pCxt, C); A = B; }
 stream_options(A) ::= stream_options(B) IGNORE EXPIRED NK_INTEGER(C).             { ((SStreamOptions*)B)->ignoreExpired = taosStr2Int8(C.z, NULL, 10); A = B; }
 stream_options(A) ::= stream_options(B) FILL_HISTORY NK_INTEGER(C).               { ((SStreamOptions*)B)->fillHistory = taosStr2Int8(C.z, NULL, 10); A = B; }
+stream_options(A) ::= stream_options(B) IGNORE UPDATE NK_INTEGER(C).              { ((SStreamOptions*)B)->ignoreUpdate = taosStr2Int8(C.z, NULL, 10); A = B; }
 
 subtable_opt(A) ::= .                                                             { A = NULL; }
 subtable_opt(A) ::= SUBTABLE NK_LP expression(B) NK_RP.                           { A = releaseRawExprNode(pCxt, B); }
@@ -1077,4 +1078,4 @@ null_ordering_opt(A) ::= NULLS LAST.
 
 %fallback ABORT AFTER ATTACH BEFORE BEGIN BITAND BITNOT BITOR BLOCKS CHANGE COMMA COMPACT CONCAT CONFLICT COPY DEFERRED DELIMITERS DETACH DIVIDE DOT EACH END FAIL 
   FILE FOR GLOB ID IMMEDIATE IMPORT INITIALLY INSTEAD ISNULL KEY MODULES NK_BITNOT NK_SEMI NOTNULL OF PLUS PRIVILEGE RAISE REPLACE RESTRICT ROW SEMI STAR STATEMENT
-  STRICT STRING TIMES UPDATE VALUES VARIABLE VIEW WAL.
+  STRICT STRING TIMES VALUES VARIABLE VIEW WAL.

source/libs/parser/src/parAstCreater.c

@@ -1763,6 +1763,7 @@ SNode* createStreamOptions(SAstCreateContext* pCxt) {
   pOptions->triggerType = STREAM_TRIGGER_AT_ONCE;
   pOptions->fillHistory = STREAM_DEFAULT_FILL_HISTORY;
   pOptions->ignoreExpired = STREAM_DEFAULT_IGNORE_EXPIRED;
+  pOptions->ignoreUpdate = STREAM_DEFAULT_IGNORE_UPDATE;
   return (SNode*)pOptions;
 }
 

source/libs/parser/src/parTokenizer.c

@@ -233,6 +233,7 @@ static SKeyword keywordTable[] = {
     {"TTL",                  TK_TTL},
     {"UNION",                TK_UNION},
     {"UNSIGNED",             TK_UNSIGNED},
+    {"UPDATE",                  TK_UPDATE},
     {"USE",                  TK_USE},
     {"USER",                 TK_USER},
     {"USERS",                TK_USERS},

source/libs/parser/src/parTranslater.c

@@ -5741,6 +5741,7 @@ static int32_t buildCreateStreamReq(STranslateContext* pCxt, SCreateStreamStmt*
     pReq->watermark = (NULL != pStmt->pOptions->pWatermark ? ((SValueNode*)pStmt->pOptions->pWatermark)->datum.i : 0);
     pReq->fillHistory = pStmt->pOptions->fillHistory;
     pReq->igExpired = pStmt->pOptions->ignoreExpired;
+    pReq->igUpdate = pStmt->pOptions->ignoreUpdate;
     columnDefNodeToField(pStmt->pTags, &pReq->pTags);
     pReq->numOfTags = LIST_LENGTH(pStmt->pTags);
   }

source/libs/parser/test/parInitialCTest.cpp

@@ -643,7 +643,8 @@ TEST_F(ParserInitialCTest, createStream) {
   auto setCreateStreamReq = [&](const char* pStream, const char* pSrcDb, const char* pSql, const char* pDstStb,
                                 int8_t igExists = 0, int8_t triggerType = STREAM_TRIGGER_AT_ONCE, int64_t maxDelay = 0,
                                 int64_t watermark = 0, int8_t igExpired = STREAM_DEFAULT_IGNORE_EXPIRED,
-                                int8_t fillHistory = STREAM_DEFAULT_FILL_HISTORY) {
+                                int8_t fillHistory = STREAM_DEFAULT_FILL_HISTORY,
+                                int8_t igUpdate = STREAM_DEFAULT_IGNORE_UPDATE) {
     snprintf(expect.name, sizeof(expect.name), "0.%s", pStream);
     snprintf(expect.sourceDB, sizeof(expect.sourceDB), "0.%s", pSrcDb);
     snprintf(expect.targetStbFullName, sizeof(expect.targetStbFullName), "0.test.%s", pDstStb);
@@ -654,6 +655,7 @@ TEST_F(ParserInitialCTest, createStream) {
     expect.watermark = watermark;
     expect.fillHistory = fillHistory;
     expect.igExpired = igExpired;
+    expect.igUpdate = igUpdate;
   };
 
   auto addTag = [&](const char* pFieldName, uint8_t type, int32_t bytes = 0) {
@@ -699,6 +701,7 @@ TEST_F(ParserInitialCTest, createStream) {
         ASSERT_EQ(pField->flags, pExpectField->flags);
       }
     }
+    ASSERT_EQ(req.igUpdate, expect.igUpdate);
     tFreeSCMCreateStreamReq(&req);
   });
 
@@ -708,12 +711,11 @@ TEST_F(ParserInitialCTest, createStream) {
 
   setCreateStreamReq(
       "s1", "test",
-      "create stream if not exists s1 trigger max_delay 20s watermark 10s ignore expired 0 fill_history 1 into st1 "
-      "as select count(*) from t1 interval(10s)",
-      "st1", 1, STREAM_TRIGGER_MAX_DELAY, 20 * MILLISECOND_PER_SECOND, 10 * MILLISECOND_PER_SECOND, 0, 1);
-  run("CREATE STREAM IF NOT EXISTS s1 TRIGGER MAX_DELAY 20s WATERMARK 10s IGNORE EXPIRED 0 FILL_HISTORY 1 INTO st1 AS "
-      "SELECT COUNT(*) "
-      "FROM t1 INTERVAL(10S)");
+      "create stream if not exists s1 trigger max_delay 20s watermark 10s ignore expired 0 fill_history 1 ignore "
+      "update 1 into st1 as select count(*) from t1 interval(10s)",
+      "st1", 1, STREAM_TRIGGER_MAX_DELAY, 20 * MILLISECOND_PER_SECOND, 10 * MILLISECOND_PER_SECOND, 0, 1, 1);
+  run("CREATE STREAM IF NOT EXISTS s1 TRIGGER MAX_DELAY 20s WATERMARK 10s IGNORE EXPIRED 0 FILL_HISTORY 1 IGNORE "
+      "UPDATE 1 INTO st1 AS SELECT COUNT(*) FROM t1 INTERVAL(10S)");
   clearCreateStreamReq();
 
   setCreateStreamReq("s1", "test",

source/libs/planner/src/planLogicCreater.c

@@ -343,6 +343,13 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
 
   pScan->node.groupAction = GROUP_ACTION_NONE;
   pScan->node.resultDataOrder = DATA_ORDER_LEVEL_IN_BLOCK;
+  if (pCxt->pPlanCxt->streamQuery) {
+    pScan->triggerType = pCxt->pPlanCxt->triggerType;
+    pScan->watermark = pCxt->pPlanCxt->watermark;
+    pScan->deleteMark = pCxt->pPlanCxt->deleteMark;
+    pScan->igExpired = pCxt->pPlanCxt->igExpired;
+    pScan->igCheckUpdate = pCxt->pPlanCxt->igCheckUpdate;
+  }
 
   // set columns to scan
   if (TSDB_CODE_SUCCESS == code) {
@@ -705,6 +712,7 @@ static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStm
     pWindow->watermark = pCxt->pPlanCxt->watermark;
     pWindow->deleteMark = pCxt->pPlanCxt->deleteMark;
     pWindow->igExpired = pCxt->pPlanCxt->igExpired;
+    pWindow->igCheckUpdate = pCxt->pPlanCxt->igCheckUpdate;
   }
   pWindow->inputTsOrder = ORDER_ASC;
   pWindow->outputTsOrder = ORDER_ASC;

source/libs/planner/src/planOptimizer.c

@@ -328,10 +328,6 @@ static void scanPathOptSetScanWin(SScanLogicNode* pScan) {
     pScan->sliding = ((SWindowLogicNode*)pParent)->sliding;
     pScan->intervalUnit = ((SWindowLogicNode*)pParent)->intervalUnit;
     pScan->slidingUnit = ((SWindowLogicNode*)pParent)->slidingUnit;
-    pScan->triggerType = ((SWindowLogicNode*)pParent)->triggerType;
-    pScan->watermark = ((SWindowLogicNode*)pParent)->watermark;
-    pScan->deleteMark = ((SWindowLogicNode*)pParent)->deleteMark;
-    pScan->igExpired = ((SWindowLogicNode*)pParent)->igExpired;
   }
 }
 

source/libs/planner/src/planPhysiCreater.c

@@ -582,6 +582,7 @@ static int32_t createTableScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubp
   pTableScan->triggerType = pScanLogicNode->triggerType;
   pTableScan->watermark = pScanLogicNode->watermark;
   pTableScan->igExpired = pScanLogicNode->igExpired;
+  pTableScan->igCheckUpdate = pScanLogicNode->igCheckUpdate;
   pTableScan->assignBlockUid = pCxt->pPlanCxt->rSmaQuery ? true : false;
 
   int32_t code = createScanPhysiNodeFinalize(pCxt, pSubplan, pScanLogicNode, (SScanPhysiNode*)pTableScan, pPhyNode);

tests/script/tsim/stream/basic3.sim

@@ -81,6 +81,15 @@ if $rows == 0 then
   return -1
 endi
 
+sleep 3000
+
+sql drop stream if exists streamd1;
+sql drop stream if exists streamd2;
+sql drop stream if exists streamd3;
+sql drop stream if exists streamd4;
+sql drop stream if exists streamd5;
+sql drop stream if exists streamd6;
+
 sql create stream streamd10 into streamd10 as select _wstart, _wend, count(*), first(ca), last(cb) as c2 from t1 interval(10s);
 
 sql desc streamd10;
@@ -100,15 +109,6 @@ if $rows == 0 then
   return -1
 endi
 
-sleep 3000
-
-sql drop stream if exists streamd1;
-sql drop stream if exists streamd2;
-sql drop stream if exists streamd3;
-sql drop stream if exists streamd4;
-sql drop stream if exists streamd5;
-sql drop stream if exists streamd6;
-
 
 _OVER:
 system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/stream/ignoreCheckUpdate.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print step 1 start
+
+sql drop stream if exists streams0;
+sql drop database if exists test;
+sql create database test  vgroups 1;
+sql use test;
+sql create table t1(ts timestamp, a int, b int , c int);
+
+print create stream streams0 trigger at_once ignore update 1 into streamt as select  _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s);
+
+sql create stream streams0 trigger at_once ignore update 1 into streamt as select  _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s);
+
+sql insert into t1 values(1648791213000,1,1,1);
+sql insert into t1 values(1648791213000,2,2,2);
+
+$loop_count = 0
+
+loop0:
+sleep 300
+sql select * from streamt order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+
+sql insert into t1 values(1648791213000,3,3,3);
+
+$loop_count = 0
+
+loop1:
+sleep 300
+sql select * from streamt order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 3 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 3 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+print step 1 end
+
+print step 2 start
+
+sql drop stream if exists streams1;
+sql drop database if exists test1;
+sql create database test1  vgroups 1;
+sql use test1;
+sql create table t1(ts timestamp, a int, b int , c int);
+
+print create stream streams1 trigger at_once ignore update 1 into streamt1 as select  _wstart c1, count(*) c2, max(b) c3 from t1 session(ts, 10s);
+
+sql create stream streams1 trigger at_once ignore update 1 into streamt1 as select  _wstart c1, count(*) c2, max(b) c3 from t1 session(ts, 10s);
+
+sql insert into t1 values(1648791213000,1,1,1);
+sql insert into t1 values(1648791213000,2,2,2);
+
+$loop_count = 0
+
+loop2:
+sleep 300
+sql select * from streamt1 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+
+sql insert into t1 values(1648791213000,3,3,3);
+
+$loop_count = 0
+
+loop3:
+
+sleep 300
+sql select * from streamt1 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 3 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 3 then
+  print =====data02=$data02
+  goto loop3
+endi
+
+print step 2 end
+
+print step 3 start
+
+sql drop stream if exists streams2;
+sql drop database if exists test2;
+sql create database test2  vgroups 1;
+sql use test2;
+sql create table t1(ts timestamp, a int, b int , c int);
+
+print create stream streams2 trigger at_once ignore update 1 into streamt2 as select  _wstart c1, count(*) c2, max(b) c3 from t1 state_window(c);
+
+sql create stream streams2 trigger at_once ignore update 1 into streamt2 as select  _wstart c1, count(*) c2, max(b) c3 from t1 state_window(c);
+
+sql insert into t1 values(1648791213000,1,1,1);
+sql insert into t1 values(1648791213000,2,2,1);
+
+$loop_count = 0
+
+loop2:
+sleep 300
+sql select * from streamt2 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop2
+endi
+
+
+sql insert into t1 values(1648791213000,3,3,1);
+
+$loop_count = 0
+
+loop3:
+
+sleep 300
+sql select * from streamt2 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 3 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 3 then
+  print =====data02=$data02
+  goto loop3
+endi
+
+print step 3 end
+
+print step 4 start
+
+sql drop stream if exists streams3;
+sql drop database if exists test3;
+sql create database test3  vgroups 4;
+sql use test3;
+sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+
+print create stream streams3 trigger at_once ignore update 1 into streamt3 as select  _wstart c1, count(*) c2, max(b) c3 from st interval(10s);
+
+sql create stream streams3 trigger at_once ignore update 1 into streamt3 as select  _wstart c1, count(*) c2, max(b) c3 from st interval(10s);
+
+sql insert into t1 values(1648791213000,1,1,1);
+sql insert into t1 values(1648791213000,2,2,2);
+
+sql insert into t2 values(1648791213000,1,1,1);
+sql insert into t2 values(1648791213000,2,2,2);
+
+$loop_count = 0
+
+loop0:
+sleep 300
+sql select * from streamt3 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 4 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+
+sql insert into t1 values(1648791213000,3,3,3);
+
+$loop_count = 0
+
+loop1:
+sleep 300
+sql select * from streamt3 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 5 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 3 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+sql insert into t2 values(1648791213000,4,4,4);
+
+$loop_count = 0
+
+loop1:
+sleep 300
+sql select * from streamt3 order by 1,2,3;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 6 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 4 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+print step 4 end
+
+system sh/stop_dnodes.sh

tests/script/tsim/stream/ignoreExpiredData.sim

@@ -52,6 +52,7 @@ sql insert into t1 values(1648791213000,1,2,3,1.0);
 sql insert into t1 values(1648791223001,1,2,3,1.1);
 sql insert into t1 values(1648791233002,2,2,3,2.1);
 sql insert into t1 values(1648791243003,2,2,3,3.1);
+sleep 300
 sql insert into t1 values(1648791200000,4,2,3,4.1);
 
 $loop_count = 0
@@ -115,6 +116,7 @@ sql create stream stream_t1 trigger at_once IGNORE EXPIRED 1 into streamtST1 as
 sql create stream stream_t2 trigger at_once IGNORE EXPIRED 1 into streamtST2 as select  _wstart, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6 from st session(ts, 10s) ;
 sql insert into ts1 values(1648791211000,1,2,3);
 sql insert into ts1 values(1648791222001,2,2,3);
+sleep 300
 sql insert into ts2 values(1648791211000,1,2,3);
 sql insert into ts2 values(1648791222001,2,2,3);