refactor: do some internal refactor.

include/libs/stream/tstream.h

@@ -132,7 +132,6 @@ typedef struct {
 // ref data block, for delete
 typedef struct {
   int8_t       type;
-  int64_t      ver;
   SSDataBlock* pBlock;
 } SStreamRefDataBlock;
 
@@ -207,8 +206,6 @@ void* streamQueueNextItem(SStreamQueue* pQueue);
 SStreamDataSubmit* streamDataSubmitNew(SPackedData* pData, int32_t type);
 void               streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit);
 
-SStreamDataSubmit* streamSubmitBlockClone(SStreamDataSubmit* pSubmit);
-
 typedef struct {
   char*              qmsg;
   void*              pExecutor;   // not applicable to encoder and decoder

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

@@ -396,8 +396,7 @@ static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t ui
   pTask->dataRange.window.skey = INT64_MIN;
   pTask->dataRange.window.ekey = 1685959190000;//taosGetTimestampMs();
 
-  mDebug("0x%x----------------window:%"PRId64" - %"PRId64, pTask->id.taskId, pTask->dataRange.window.skey, pTask->dataRange.window.ekey);
-
+  mDebug("s-task:0x%x set time window:%"PRId64" - %"PRId64, pTask->id.taskId, pTask->dataRange.window.skey, pTask->dataRange.window.ekey);
 
   // all the source tasks dispatch result to a single agg node.
   setFixedDownstreamEpInfo(pTask, pDownstreamTask);

source/libs/stream/src/stream.c

@@ -17,7 +17,7 @@
 #include "ttimer.h"
 
 #define STREAM_TASK_INPUT_QUEUE_CAPACITY          20480
-#define STREAM_TASK_INPUT_QUEUE_CAPACITY_IN_SIZE  (50)
+#define STREAM_TASK_INPUT_QUEUE_CAPACITY_IN_SIZE  (30)
 #define ONE_MB_F                                  (1048576.0)
 #define QUEUE_MEM_SIZE_IN_MB(_q)                  (taosQueueMemorySize(_q) / ONE_MB_F)
 

source/libs/stream/src/streamData.c

@@ -164,26 +164,6 @@ int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubm
   return 0;
 }
 
-static FORCE_INLINE void streamDataSubmitRefInc(SStreamDataSubmit* pDataSubmit) {
-  atomic_add_fetch_32(pDataSubmit->dataRef, 1);
-}
-
-SStreamDataSubmit* streamSubmitBlockClone(SStreamDataSubmit* pSubmit) {
-  int32_t len = 0;
-  if (pSubmit->type == STREAM_INPUT__DATA_SUBMIT) {
-    len = pSubmit->submit.msgLen;
-  }
-
-  SStreamDataSubmit* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, len);
-  if (pSubmitClone == NULL) {
-    return NULL;
-  }
-
-  streamDataSubmitRefInc(pSubmit);
-  memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit));
-  return pSubmitClone;
-}
-
 SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem) {
   if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) {
     SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;