Merge pull request #18625 from taosdata/feature/stream

fix: compatility after update stream

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

@@ -647,7 +647,7 @@ typedef struct {
 } SStreamObj;
 
 int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
-int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj);
+int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj, int32_t sver);
 void    tFreeStreamObj(SStreamObj* pObj);
 
 typedef struct {

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

@@ -83,7 +83,7 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
   return pEncoder->pos;
 }
 
-int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
+int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {
   if (tStartDecode(pDecoder) < 0) return -1;
   if (tDecodeCStrTo(pDecoder, pObj->name) < 0) return -1;
 
@@ -143,8 +143,9 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
   if (tDecodeSSchemaWrapper(pDecoder, &pObj->outputSchema) < 0) return -1;
 
   // 3.0.20
-  if (tDecodeI64(pDecoder, &pObj->checkpointFreq) < 0) return -1;
-
+  if (sver >= 2) {
+    if (tDecodeI64(pDecoder, &pObj->checkpointFreq) < 0) return -1;
+  }
   tEndDecode(pDecoder);
   return 0;
 }

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

@@ -28,7 +28,7 @@
 #include "parser.h"
 #include "tname.h"
 
-#define MND_STREAM_VER_NUMBER   1
+#define MND_STREAM_VER_NUMBER   2
 #define MND_STREAM_RESERVE_SIZE 64
 
 static int32_t mndStreamActionInsert(SSdb *pSdb, SStreamObj *pStream);
@@ -133,7 +133,7 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
   int8_t sver = 0;
   if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto STREAM_DECODE_OVER;
 
-  if (sver != MND_STREAM_VER_NUMBER) {
+  if (sver != 1 && sver != 2) {
     terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
     goto STREAM_DECODE_OVER;
   }
@@ -153,7 +153,7 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
 
   SDecoder decoder;
   tDecoderInit(&decoder, buf, tlen + 1);
-  if (tDecodeSStreamObj(&decoder, pStream) < 0) {
+  if (tDecodeSStreamObj(&decoder, pStream, sver) < 0) {
     tDecoderClear(&decoder);
     goto STREAM_DECODE_OVER;
   }