taosdata · guanshengliang · Nov 13, 2025 · Oct 30, 2025 · Oct 30, 2025 · Oct 30, 2025
@@ -331,7 +331,7 @@ Note: Users with SYSINFO property set to 0 cannot view this table.
 |:----|:-----------|:-----------|:--------------------|
 | 1   | db_name    | VARCHAR(32) | Database name                              |
 | 2   | vgroup_id  | INT         | vgroup ID                                  |
-| 3   | wal        | BIGINT      | WAL file size, in KB                       |
+| 3   | wal_size   | BIGINT      | WAL file size, in KB                       |
 | 4   | data1      | BIGINT      | Data file size on primary storage, in KB   |
 | 5   | data2      | BIGINT      | Data file size on secondary storage, in KB |
 | 6   | data3      | BIGINT      | Data file size on tertiary storage, in KB  |

@@ -335,7 +335,7 @@ TDengine 内置了一个名为 `INFORMATION_SCHEMA` 的数据库，提供对数
 |:----|:-----------|:------------|:--------|
 | 1   | db_name    | VARCHAR(32) | 数据库名称                      |
 | 2   | vgroup_id  | INT         | vgroup 的 ID                   |
-| 3   | wal        | BIGINT      | wal 文件大小，单位为 KB           |
+| 3   | wal_size   | BIGINT      | wal 文件大小，单位为 KB           |
 | 4   | data1      | BIGINT      | 一级存储上数据文件的大小，单位为 KB   |
 | 5   | data2      | BIGINT      | 二级存储上数据文件的大小，单位为 KB  |
 | 6   | data3      | BIGINT      | 三级存储上数据文件的大小，单位为 KB  |

@@ -333,6 +333,7 @@ typedef enum ENodeType {
   QUERY_NODE_REVOKE_STMT,
   QUERY_NODE_ALTER_CLUSTER_STMT,
   QUERY_NODE_S3MIGRATE_DATABASE_STMT,
+  QUERY_NODE_TRIM_DATABASE_WAL_STMT,
   QUERY_NODE_CREATE_TSMA_STMT,
   QUERY_NODE_DROP_TSMA_STMT,
   QUERY_NODE_CREATE_VIRTUAL_TABLE_STMT,
@@ -375,6 +376,7 @@ typedef enum ENodeType {
   QUERY_NODE_ASSIGN_LEADER_STMT,
   QUERY_NODE_SHOW_CREATE_TSMA_STMT,
   QUERY_NODE_SHOW_CREATE_VTABLE_STMT,
+  QUERY_NODE_SET_VGROUP_KEEP_VERSION_STMT,
 
   // show statement nodes
   // see 'sysTableShowAdapter', 'SYSTABLE_SHOW_TYPE_OFFSET'
@@ -1711,6 +1713,14 @@ typedef struct {
 int32_t tSerializeSVS3MigrateDbReq(void* buf, int32_t bufLen, SVS3MigrateDbReq* pReq);
 int32_t tDeserializeSVS3MigrateDbReq(void* buf, int32_t bufLen, SVS3MigrateDbReq* pReq);
 
+typedef struct {
+  int32_t vgId;
+  int64_t keepVersion;
+} SMndSetVgroupKeepVersionReq;
+
+int32_t tSerializeSMndSetVgroupKeepVersionReq(void* buf, int32_t bufLen, SMndSetVgroupKeepVersionReq* pReq);
+int32_t tDeserializeSMndSetVgroupKeepVersionReq(void* buf, int32_t bufLen, SMndSetVgroupKeepVersionReq* pReq);
+
 typedef struct {
   int32_t timestampSec;
   int32_t ttlDropMaxCount;
@@ -4094,6 +4104,13 @@ int32_t tSerializeSMDropStreamReq(void* buf, int32_t bufLen, const SMDropStreamR
 int32_t tDeserializeSMDropStreamReq(void* buf, int32_t bufLen, SMDropStreamReq* pReq);
 void    tFreeMDropStreamReq(SMDropStreamReq* pReq);
 
+typedef struct SVndSetKeepVersionReq {
+  int64_t keepVersion;
+} SVndSetKeepVersionReq;
+
+int32_t tSerializeSVndSetKeepVersionReq(void* buf, int32_t bufLen, SVndSetKeepVersionReq* pReq);
+int32_t tDeserializeSVndSetKeepVersionReq(void* buf, int32_t bufLen, SVndSetKeepVersionReq* pReq);
+
 typedef struct SVUpdateCheckpointInfoReq {
   SMsgHead head;
   int64_t  streamId;

@@ -135,6 +135,8 @@
   TD_DEF_MSG_TYPE(TDMT_MND_GET_DB_INFO, "get-db-info", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_USER_WHITELIST_DUAL, "get-user-whitelist-dual", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_RETRIEVE_IP_WHITE_DUAL, "retrieve-ip-white-dual", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_SET_VGROUP_KEEP_VERSION, "set-vgroup-keep-version", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TRIM_DB_WAL, "trim-db-wal", NULL, NULL)
   TD_CLOSE_MSG_SEG(TDMT_DND_MSG)
 
   TD_NEW_MSG_SEG(TDMT_MND_MSG)  // 1<<8
@@ -327,6 +329,8 @@
   TD_DEF_MSG_TYPE(TDMT_VND_TABLE_NAME, "vnode-table-name", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_VSUBTABLES_META, "vnode-virtual_stables-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_VSTB_REF_DBS, "vnode-virtual-stables-ref-dbs", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_SET_KEEP_VERSION, "vnode-set-keep-version", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_TRIM_WAL, "vnode-trim-wal", NULL, NULL)
   TD_CLOSE_MSG_SEG(TDMT_VND_MSG)
 
   TD_NEW_MSG_SEG(TDMT_SCH_MSG)  // 3<<8

@@ -742,6 +742,17 @@ typedef struct SBalanceVgroupLeaderStmt {
   char      dbName[TSDB_DB_NAME_LEN];
 } SBalanceVgroupLeaderStmt;
 
+typedef struct SSetVgroupKeepVersionStmt {
+  ENodeType type;
+  int32_t   vgId;
+  int64_t   keepVersion;
+} SSetVgroupKeepVersionStmt;
+
+typedef struct STrimDbWalStmt {
+  ENodeType type;
+  char      dbName[TSDB_DB_FNAME_LEN];
+} STrimDbWalStmt;
+
 typedef struct SMergeVgroupStmt {
   ENodeType type;
   int32_t   vgId1;

@@ -281,7 +281,7 @@ int64_t   syncGetTerm(int64_t rid);
 int32_t   syncProcessMsg(int64_t rid, SRpcMsg* pMsg);
 int32_t   syncReconfig(int64_t rid, SSyncCfg* pCfg);
 int32_t   syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex);
-int32_t   syncEndSnapshot(int64_t rid);
+int32_t   syncEndSnapshot(int64_t rid, bool forceTrim);
 int32_t   syncLeaderTransfer(int64_t rid);
 int32_t   syncStepDown(int64_t rid, SyncTerm newTerm);
 bool      syncIsReadyForRead(int64_t rid);

@@ -119,6 +119,8 @@ typedef struct SWal {
   TdThreadRwlock mutex;
   // ref
   SHashObj *pRefHash;  // refId -> SWalRef
+  // keep version for preventing auto deletion
+  int64_t keepVersion;
   // path
   char path[WAL_PATH_LEN];
 
@@ -180,7 +182,7 @@ int32_t walCommit(SWal *, int64_t ver);
 int32_t walRollback(SWal *, int64_t ver);
 // notify that previous logs can be pruned safely
 int32_t walBeginSnapshot(SWal *, int64_t ver, int64_t logRetention);
-int32_t walEndSnapshot(SWal *);
+int32_t walEndSnapshot(SWal *, bool forceTrim);
 int32_t walRestoreFromSnapshot(SWal *, int64_t ver);
 void    walApplyVer(SWal *, int64_t ver);
 
@@ -219,6 +221,7 @@ int64_t walGetLastVer(SWal *);
 int64_t walGetVerRetention(SWal *pWal, int64_t bytes);
 int64_t walGetCommittedVer(SWal *);
 int64_t walGetAppliedVer(SWal *);
+int32_t walSetKeepVersion(SWal *pWal, int64_t ver);
 
 #ifdef __cplusplus
 }

@@ -88,6 +88,7 @@ struct tmq_conf_t {
   char           clientId[TSDB_CLIENT_ID_LEN];
   char           groupId[TSDB_CGROUP_LEN];
   int8_t         autoCommit;
+  int8_t         enableWalMarker;
   int8_t         resetOffset;
   int8_t         withTbName;
   int8_t         snapEnable;
@@ -118,6 +119,7 @@ struct tmq_t {
   int8_t         withTbName;
   int8_t         useSnapshot;
   int8_t         autoCommit;
+  int8_t         enableWalMarker;
   int32_t        autoCommitInterval;
   int32_t        sessionTimeoutMs;
   int32_t        heartBeatIntervalMs;
@@ -304,6 +306,7 @@ tmq_conf_t* tmq_conf_new() {
 
   conf->withTbName = false;
   conf->autoCommit = true;
+  conf->enableWalMarker = false;
   conf->autoCommitInterval = DEFAULT_AUTO_COMMIT_INTERVAL;
   conf->resetOffset = TMQ_OFFSET__RESET_LATEST;
   conf->enableBatchMeta = false;
@@ -364,6 +367,19 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
     }
   }
 
+  if (strcasecmp(key, "enable.wal.marker") == 0) {
+    if (strcasecmp(value, "true") == 0) {
+      conf->enableWalMarker = true;
+      return TMQ_CONF_OK;
+    } else if (strcasecmp(value, "false") == 0) {
+      conf->enableWalMarker = false;
+      return TMQ_CONF_OK;
+    } else {
+      tqErrorC("invalid value for enable.wal.marker:%s", value);
+      return TMQ_CONF_INVALID;
+    }
+  }
+
   if (strcasecmp(key, "auto.commit.interval.ms") == 0) {
     int64_t tmp;
     code = taosStr2int64(value, &tmp);
@@ -659,6 +675,7 @@ static int32_t doSendCommitMsg(tmq_t* tmq, int32_t vgId, SEpSet* epSet, STqOffse
   SMqVgOffset pOffset = {0};
 
   pOffset.consumerId = tmq->consumerId;
+  // pOffset.markWal = tmq->enableWalMarker;
   pOffset.offset.val = *offset;
   (void)snprintf(pOffset.offset.subKey, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", tmq->groupId, TMQ_SEPARATOR, pTopicName);
   int32_t len = 0;
@@ -785,6 +802,52 @@ static int32_t getClientVg(tmq_t* tmq, char* pTopicName, int32_t vgId, SMqClient
   return *pVg == NULL ? TSDB_CODE_TMQ_INVALID_VGID : TSDB_CODE_SUCCESS;
 }
 
+static int32_t sendWalMarkMsgToMnodeCb(void* param, SDataBuf* pMsg, int32_t code) {
+  if (pMsg) {
+    taosMemoryFreeClear(pMsg->pEpSet);
+    taosMemoryFreeClear(pMsg->pData);
+  }
+  tqDebugC("sendWalMarkMsgToMnodeCb code:%d", code);
+  return 0;
+}
+
+static void asyncSendWalMarkMsgToMnode(tmq_t* tmq, int32_t vgId, int64_t keepVersion) {
+  if (tmq == NULL) return ;
+  void*           buf = NULL;
+  SMsgSendInfo*   sendInfo = NULL;
+  SMndSetVgroupKeepVersionReq req = {0};
+
+  tqDebugC("consumer:0x%" PRIx64 " send vgId:%d keepVersion:%"PRId64, tmq->consumerId, vgId, keepVersion);
+  req.vgId = vgId;
+  req.keepVersion = keepVersion;
+
+  int32_t tlen = tSerializeSMndSetVgroupKeepVersionReq(NULL, 0, &req);
+  buf = taosMemoryMalloc(tlen);
+  if (buf == NULL) {
+    return;
+  }
+  tlen = tSerializeSMndSetVgroupKeepVersionReq(buf, tlen, &req);
+
+  sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
+  if (sendInfo == NULL) {
+    taosMemoryFree(buf);
+    return;
+  }
+
+  sendInfo->msgInfo = (SDataBuf){.pData = buf, .len = tlen, .handle = NULL};
+  sendInfo->requestId = generateRequestId();
+  sendInfo->fp = sendWalMarkMsgToMnodeCb;
+  sendInfo->msgType = TDMT_MND_SET_VGROUP_KEEP_VERSION;
+
+  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
+
+  int32_t code = asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, NULL, sendInfo);
+  if (code != 0) {
+    tqErrorC("consumer:0x%" PRIx64 " send wal mark msg to mnode failed, code:%s", tmq->consumerId,
+             tstrerror(terrno));
+  }
+}
+
 static int32_t innerCommit(tmq_t* tmq, char* pTopicName, STqOffsetVal* offsetVal, SMqClientVg* pVg, SMqCommitCbParamSet* pParamSet){
   if (tmq == NULL || pTopicName == NULL || offsetVal == NULL || pVg == NULL || pParamSet == NULL) {
     return TSDB_CODE_INVALID_PARA;
@@ -811,6 +874,9 @@ static int32_t innerCommit(tmq_t* tmq, char* pTopicName, STqOffsetVal* offsetVal
     return code;
   }
 
+  if (tmq->enableWalMarker && offsetVal->type == TMQ_OFFSET__LOG) {
+    asyncSendWalMarkMsgToMnode(tmq, pVg->vgId, offsetVal->version);
+  }
   tqDebugC("consumer:0x%" PRIx64 " topic:%s on vgId:%d send commit msg success, send offset:%s committed:%s",
            tmq->consumerId, pTopicName, pVg->vgId, offsetBuf, commitBuf);
   tOffsetCopy(&pVg->offsetInfo.committedOffset, offsetVal);
@@ -1538,6 +1604,7 @@ void tmqClearUnhandleMsg(tmq_t* tmq) {
 int32_t tmqSubscribeCb(void* param, SDataBuf* pMsg, int32_t code) {
   if (pMsg) {
     taosMemoryFreeClear(pMsg->pEpSet);
+    taosMemoryFreeClear(pMsg->pData);
   }
 
   if (param == NULL) {
@@ -1739,6 +1806,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   tstrncpy(pTmq->groupId, conf->groupId, TSDB_CGROUP_LEN);
   pTmq->withTbName = conf->withTbName;
   pTmq->useSnapshot = conf->snapEnable;
+  pTmq->enableWalMarker = conf->enableWalMarker;
   pTmq->autoCommit = conf->autoCommit;
   pTmq->autoCommitInterval = conf->autoCommitInterval;
   pTmq->sessionTimeoutMs = conf->sessionTimeoutMs;

@@ -6085,6 +6085,42 @@ int32_t tDeserializeSTrimDbReq(void *buf, int32_t bufLen, STrimDbReq *pReq) {
   TAOS_CHECK_EXIT(tStartDecode(&decoder));
   TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pReq->db));
   TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->maxSpeed));
+  tEndDecode(&decoder); 
+
+_exit:
+  tDecoderClear(&decoder);
+  return code;
+}
+
+int32_t tSerializeSVndSetKeepVersionReq(void *buf, int32_t bufLen, SVndSetKeepVersionReq *pReq) {
+  SEncoder encoder = {0};
+  int32_t  code = 0;
+  int32_t  lino;
+  int32_t  tlen;
+  tEncoderInit(&encoder, buf, bufLen);
+
+  TAOS_CHECK_EXIT(tStartEncode(&encoder));
+  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->keepVersion));
+  tEndEncode(&encoder);
+
+_exit:
+  if (code) {
+    tlen = code;
+  } else {
+    tlen = encoder.pos;
+  }
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSVndSetKeepVersionReq(void *buf, int32_t bufLen, SVndSetKeepVersionReq *pReq) {
+  SDecoder decoder = {0};
+  int32_t  code = 0;
+  int32_t  lino;
+  tDecoderInit(&decoder, buf, bufLen);
+
+  TAOS_CHECK_EXIT(tStartDecode(&decoder));
+  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->keepVersion));
   tEndDecode(&decoder);
 
 _exit:
@@ -6202,6 +6238,45 @@ int32_t tDeserializeSVS3MigrateDbReq(void *buf, int32_t bufLen, SVS3MigrateDbReq
   return code;
 }
 
+int32_t tSerializeSMndSetVgroupKeepVersionReq(void *buf, int32_t bufLen, SMndSetVgroupKeepVersionReq *pReq) {
+  SEncoder encoder = {0};
+  int32_t  code = 0;
+  int32_t  lino;
+  int32_t  tlen;
+  tEncoderInit(&encoder, buf, bufLen);
+
+  TAOS_CHECK_EXIT(tStartEncode(&encoder));
+  TAOS_CHECK_EXIT(tEncodeI32(&encoder, pReq->vgId));
+  TAOS_CHECK_EXIT(tEncodeI64(&encoder, pReq->keepVersion));
+  tEndEncode(&encoder);
+
+_exit:
+  if (code) {
+    tlen = code;
+  } else {
+    tlen = encoder.pos;
+  }
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSMndSetVgroupKeepVersionReq(void *buf, int32_t bufLen, SMndSetVgroupKeepVersionReq *pReq) {
+  SDecoder decoder = {0};
+  int32_t  code = 0;
+  int32_t  lino;
+  tDecoderInit(&decoder, buf, bufLen);
+
+  TAOS_CHECK_EXIT(tStartDecode(&decoder));
+  TAOS_CHECK_EXIT(tDecodeI32(&decoder, &pReq->vgId));
+  TAOS_CHECK_EXIT(tDecodeI64(&decoder, &pReq->keepVersion));
+
+  tEndDecode(&decoder);
+
+_exit:
+  tDecoderClear(&decoder);
+  return code;
+}
+
 int32_t tSerializeSVDropTtlTableReq(void *buf, int32_t bufLen, SVDropTtlTableReq *pReq) {
   SEncoder encoder = {0};
   int32_t  code = 0;
@@ -12552,6 +12627,7 @@ int32_t tDecodeSTqOffset(SDecoder *pDecoder, STqOffset *pOffset) {
 int32_t tEncodeMqVgOffset(SEncoder *pEncoder, const SMqVgOffset *pOffset) {
   TAOS_CHECK_RETURN(tEncodeSTqOffset(pEncoder, &pOffset->offset));
   TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pOffset->consumerId));
+  // TAOS_CHECK_RETURN(tEncodeI8(pEncoder, pOffset->markWal));
   return 0;
 }
 

@@ -311,6 +311,8 @@ static const SSysDbTableSchema vgroupsSchema[] = {
     {.name = "cacheload", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "cacheelements", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "tsma", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},
+    {.name = "keep_version", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = true},
+    {.name = "keep_version_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = true},
     // {.name = "compact_start_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
 };
 
@@ -501,7 +503,7 @@ static const SSysDbTableSchema encryptionsSchema[] = {
 static const SSysDbTableSchema diskUsageSchema[] = {
     {.name = "db_name", .bytes = 32 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
-    {.name = "wal", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
+    {.name = "wal_size", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
     {.name = "data1", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
     {.name = "data2", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
     {.name = "data3", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},