From e03deeaed80edcacb458be8d783e9c6937932233 Mon Sep 17 00:00:00 2001
From: 54liuyao <54liuyao@163.com>
Date: Tue, 17 Jan 2023 16:21:51 +0800
Subject: [PATCH] feat:add stream delete mark

---
 include/libs/stream/streamState.h             |  1 +
 source/dnode/mnode/impl/src/mndStream.c       |  2 +
 source/libs/executor/src/timewindowoperator.c |  6 +++
 source/libs/stream/src/streamState.c          | 43 +++++++++++++++++++
 4 files changed, 52 insertions(+)

diff --git a/include/libs/stream/streamState.h b/include/libs/stream/streamState.h
index 8fdac0da7f..912c09a0fb 100644
--- a/include/libs/stream/streamState.h
+++ b/include/libs/stream/streamState.h
@@ -110,6 +110,7 @@ int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal
 
 #if 0
 char* streamStateSessionDump(SStreamState* pState);
+char* streamStateIntervalDump(SStreamState* pState);
 #endif
 
 #ifdef __cplusplus
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 83ed6eea78..38d4e5e1f7 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -295,6 +295,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
   pObj->triggerParam = pCreate->maxDelay;
   pObj->watermark = pCreate->watermark;
   pObj->fillHistory = pCreate->fillHistory;
+  pObj->deleteMark = pCreate->deleteMark;
 
   memcpy(pObj->sourceDb, pCreate->sourceDB, TSDB_DB_FNAME_LEN);
   SDbObj *pSourceDb = mndAcquireDb(pMnode, pCreate->sourceDB);
@@ -343,6 +344,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,
+      .deleteMark = pObj->deleteMark,
   };
 
   // using ast and param to build physical plan
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index 449c52d77f..c5cd64b493 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -4803,6 +4803,12 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
   taosHashCleanup(pUpdatedMap);
 
+#if 0
+  char* pBuf = streamStateIntervalDump(pInfo->pState);
+  qDebug("===stream===interval state%s", pBuf);
+  taosMemoryFree(pBuf);
+#endif
+
   doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
   if (pInfo->pDelRes->info.rows > 0) {
     printDataBlock(pInfo->pDelRes, "single interval delete");
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index 1952d9ab52..b890d145e5 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -879,4 +879,47 @@ char* streamStateSessionDump(SStreamState* pState) {
   streamStateFreeCur(pCur);
   return dumpBuf;
 }
+
+char* streamStateIntervalDump(SStreamState* pState) {
+  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
+  if (pCur == NULL) {
+    return NULL;
+  }
+  pCur->number = pState->number;
+  if (tdbTbcOpen(pState->pTdbState->pStateDb, &pCur->pCur, NULL) < 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+  tdbTbcMoveToFirst(pCur->pCur);
+
+  SWinKey key = {0};
+  void*       buf = NULL;
+  int32_t     bufSize = 0;
+  int32_t     code = streamStateGetKVByCur(pCur, &key, (const void **)&buf, &bufSize);
+  if (code != 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+
+  int32_t size = 2048;
+  char*   dumpBuf = taosMemoryCalloc(size, 1);
+  int64_t len = 0;
+  len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.ts);
+  // len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
+  len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
+  while (1) {
+    tdbTbcMoveToNext(pCur->pCur);
+    key = (SWinKey){0};
+    code = streamStateGetKVByCur(pCur, &key, NULL, 0);
+    if (code != 0) {
+      streamStateFreeCur(pCur);
+      return dumpBuf;
+    }
+    len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.ts);
+    // len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
+    len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
+  }
+  streamStateFreeCur(pCur);
+  return dumpBuf;
+}
 #endif