Merge remote-tracking branch 'origin/enh/triggerCheckPoint2' into enh/triggerCheckPoint2

include/libs/stream/tstream.h

@@ -382,8 +382,11 @@ SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, int8_t fillHisto
 int32_t      tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
 int32_t      tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
 void         tFreeStreamTask(SStreamTask* pTask);
-int32_t      tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem);
-bool         tInputQueueIsFull(const SStreamTask* pTask);
+
+int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo);
+
+int32_t tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem);
+bool    tInputQueueIsFull(const SStreamTask* pTask);
 
 typedef struct {
   SMsgHead head;

source/libs/stream/src/streamMeta.c

@@ -438,18 +438,14 @@ int64_t streamGetLatestCheckpointId(SStreamMeta* pMeta) {
 
   tdbTbcMoveToFirst(pCur);
   while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
-    SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
-    if (pTask == NULL) {
-      goto _err;
-    }
-
+    SCheckpointInfo info;
     tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
-    tDecodeStreamTask(&decoder, pTask);
+    if (tDecodeStreamTaskChkInfo(&decoder, &info) < 0) {
+      continue;
+    }
     tDecoderClear(&decoder);
 
-    chkpId = TMAX(chkpId, pTask->chkInfo.checkpointId);
-
-    taosMemoryFree(pTask);  // fix mem leak later
+    chkpId = TMAX(chkpId, info.checkpointId);
   }
 
 _err:

source/libs/stream/src/streamSnapshot.c

@@ -219,6 +219,7 @@ int32_t streamSnapReaderOpen(void* pMeta, int64_t sver, int64_t ever, char* path
   }
   // const char* path = NULL;
   if (streamSnapHandleInit(&pReader->handle, (char*)path) < 0) {
+    taosMemoryFree(pReader);
     return -1;
   }
 

source/libs/stream/src/streamTask.c

@@ -132,6 +132,35 @@ int32_t tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
   return pEncoder->pos;
 }
 
+int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo) {
+  int64_t skip64;
+  int8_t  skip8;
+  int32_t skip32;
+  int16_t skip16;
+  SEpSet  epSet;
+
+  if (tStartDecode(pDecoder) < 0) return -1;
+  if (tDecodeI64(pDecoder, &skip64) < 0) return -1;
+  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
+  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
+  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;
+  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;
+  if (tDecodeI16(pDecoder, &skip16) < 0) return -1;
+
+  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;
+  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;
+
+  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
+  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
+  if (tDecodeSEpSet(pDecoder, &epSet) < 0) return -1;
+
+  if (tDecodeI64(pDecoder, &pChkpInfo->checkpointId) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pChkpInfo->checkpointVer) < 0) return -1;
+
+  tEndDecode(pDecoder);
+  return 0;
+}
+
 int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
   if (tStartDecode(pDecoder) < 0) return -1;
   if (tDecodeI64(pDecoder, &pTask->id.streamId) < 0) return -1;
@@ -236,7 +265,7 @@ void tFreeStreamTask(SStreamTask* pTask) {
     tSimpleHashCleanup(pTask->tbSink.pTblInfo);
   } else if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
     taosArrayDestroy(pTask->shuffleDispatcher.dbInfo.pVgroupInfos);
-    pTask->checkReqIds =taosArrayDestroy(pTask->checkReqIds);
+    pTask->checkReqIds = taosArrayDestroy(pTask->checkReqIds);
   }
 
   if (pTask->pState) {

tests/script/tsim/stream/checkpointInterval0.sim

@@ -14,6 +14,7 @@ sql use test;
 
 sql create table t1(ts timestamp, a int, b int , c int, d double);
 sql create stream streams0 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0   into streamt as select  _wstart, count(*) c1, sum(a) from t1 interval(10s);
+sql create stream streams1 trigger window_close IGNORE EXPIRED 0 IGNORE UPDATE 0   into streamt1 as select  _wstart, count(*) c1, sum(a) from t1 interval(10s);
 sql insert into t1 values(1648791213000,1,2,3,1.0);
 sql insert into t1 values(1648791213001,2,2,3,1.1);
 
@@ -45,6 +46,23 @@ if $data02 != 3 then
   goto loop0
 endi
 
+$loop_count = 0
+
+loop01:
+sleep 1000
+
+sql select * from streamt1;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 0 then
+  print =====rows=$rows  expect 1
+  goto loop01
+endi
+
 print waiting for checkpoint generation 1 ......
 
 sleep 25000
@@ -126,6 +144,36 @@ if $data12 != 4 then
   goto loop2
 endi
 
+
+$loop_count = 0
+
+loop3:
+sleep 1000
+
+print select * from streamt1;
+sql select * from streamt1;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 1 then
+  print =====rows=$rows  expect 2
+  goto loop3
+endi
+
+# row 0
+if $data01 != 3 then
+  print =====data01=$data01
+  goto loop3
+endi
+
+if $data02 != 6 then
+  print =====data02=$data02
+  goto loop3
+endi
+
 print step 2
 
 print restart taosd 02 ......
@@ -136,7 +184,7 @@ system sh/exec.sh -n dnode1 -s start
 
 sql insert into t1 values(1648791223004,5,2,3,1.1);
 
-loop20:
+loop4:
 sleep 1000
 
 sql select * from streamt;
@@ -148,29 +196,58 @@ endi
 
 if $rows != 2 then
   print =====rows=$rows  expect 2
-  goto loop20
+  goto loop4
 endi
 
 # row 0
 if $data01 != 3 then
   print =====data01=$data01
-  goto loop20
+  goto loop4
 endi
 
 if $data02 != 6 then
   print =====data02=$data02
-  goto loop20
+  goto loop4
 endi
 
 # row 1
 if $data11 != 2 then
   print =====data11=$data11
-  goto loop20
+  goto loop4
 endi
 
 if $data12 != 9 then
   print =====data12=$data12
-  goto loop20
+  goto loop4
+endi
+
+$loop_count = 0
+
+loop5:
+sleep 1000
+
+print select * from streamt1;
+sql select * from streamt1;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $rows != 1 then
+  print =====rows=$rows  expect 2
+  goto loop5
+endi
+
+# row 0
+if $data01 != 3 then
+  print =====data01=$data01
+  goto loop5
+endi
+
+if $data02 != 6 then
+  print =====data02=$data02
+  goto loop5
 endi
 
 print end---------------------------------