From 77f287d9be123bd1206882097d774991680fd2e9 Mon Sep 17 00:00:00 2001
From: yihaoDeng <luomoxyz@126.com>
Date: Thu, 8 Jun 2023 07:19:59 +0000
Subject: [PATCH] add trigger checkpoint

---
 include/libs/stream/tstream.h         | 1 -
 source/common/src/tglobal.c           | 2 +-
 source/dnode/vnode/src/vnd/vnodeSvr.c | 5 +++++
 3 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 8316e6ef50..c4fa29747c 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -346,7 +346,6 @@ typedef struct SStreamMeta {
   void*        streamBackend;
   int32_t      streamBackendId;
   int64_t      streamBackendRid;
-  SHashObj*    pTaskBackendUnique;
 } SStreamMeta;
 
 int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 007910754c..f60a823b21 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -200,7 +200,7 @@ int64_t tsWalFsyncDataSizeLimit = (100 * 1024 * 1024L);
 // internal
 int32_t tsTransPullupInterval = 2;
 int32_t tsMqRebalanceInterval = 2;
-int32_t tsStreamCheckpointTickInterval = 50;
+int32_t tsStreamCheckpointTickInterval = 120;
 int32_t tsTtlUnit = 86400;
 int32_t tsTtlPushInterval = 3600;
 int32_t tsGrantHBInterval = 60;
diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c
index c3fb5e5ad4..552ee48978 100644
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
@@ -435,12 +435,17 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg
         goto _err;
       }
     } break;
+
+    case TDMT_VND_STREAM_CHECK_POINT_SOURCE: {
+      if (tqProcessStreamCheckPointReq(pVnode->pTq, ver, pReq, len) < 0) goto _err;
+    } break;
     case TDMT_VND_ALTER_CONFIRM:
       needCommit = pVnode->config.hashChange;
       if (vnodeProcessAlterConfirmReq(pVnode, ver, pReq, len, pRsp) < 0) {
         goto _err;
       }
       break;
+
     case TDMT_VND_ALTER_CONFIG:
       vnodeProcessAlterConfigReq(pVnode, ver, pReq, len, pRsp);
       break;