Merge branch 'refact/fillhistory1' into refact/fillhistory

cmake/rocksdb_CMakeLists.txt.in

 
 # rocksdb
 ExternalProject_Add(rocksdb
-        GIT_REPOSITORY https://github.com/facebook/rocksdb.git
-        GIT_TAG v8.1.1
-        SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb"
-        CONFIGURE_COMMAND ""
-        BUILD_COMMAND ""
-        INSTALL_COMMAND ""
-        TEST_COMMAND ""
+	URL https://github.com/facebook/rocksdb/archive/refs/tags/v8.1.1.tar.gz
+  	URL_HASH MD5=3b4c97ee45df9c8a5517308d31ab008b
+  	DOWNLOAD_NO_PROGRESS 1
+	DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download" 
+	SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb" 
+    	CONFIGURE_COMMAND ""
+    	BUILD_COMMAND ""
+    	INSTALL_COMMAND ""
+    	TEST_COMMAND ""
         )
+

include/libs/function/function.h

@@ -129,30 +129,38 @@ typedef struct SSerializeDataHandle {
 } SSerializeDataHandle;
 
 // incremental state storage
+
+typedef struct SBackendCfWrapper {
+  void          *rocksdb;
+  void         **pHandle;
+  void          *writeOpts;
+  void          *readOpts;
+  void         **cfOpts;
+  void          *dbOpt;
+  void          *param;
+  void          *env;
+  SListNode     *pComparNode;
+  void          *pBackend;
+  void          *compactFactory;
+  TdThreadRwlock rwLock;
+  bool           remove;
+  int64_t        backendId;
+  char           idstr[64];
+} SBackendCfWrapper;
 typedef struct STdbState {
-  void               *rocksdb;
-  void              **pHandle;
-  void               *writeOpts;
-  void               *readOpts;
-  void              **cfOpts;
-  void               *dbOpt;
+  SBackendCfWrapper *pBackendCfWrapper;
+  int64_t            backendCfWrapperId;
+  char               idstr[64];
+
   struct SStreamTask *pOwner;
-  void               *param;
-  void               *env;
-  SListNode          *pComparNode;
-  void               *pBackend;
-  char                idstr[64];
-  void               *compactFactory;
-  TdThreadRwlock      rwLock;
-
-  void *db;
-  void *pStateDb;
-  void *pFuncStateDb;
-  void *pFillStateDb;  // todo refactor
-  void *pSessionStateDb;
-  void *pParNameDb;
-  void *pParTagDb;
-  void *txn;
+  void               *db;
+  void               *pStateDb;
+  void               *pFuncStateDb;
+  void               *pFillStateDb;  // todo refactor
+  void               *pSessionStateDb;
+  void               *pParNameDb;
+  void               *pParTagDb;
+  void               *txn;
 } STdbState;
 
 typedef struct {

include/libs/stream/tstream.h

@@ -44,8 +44,8 @@ enum {
   TASK_STATUS__DROPPING,
   TASK_STATUS__FAIL,
   TASK_STATUS__STOP,
-  TASK_STATUS__SCAN_HISTORY, // stream task scan history data by using tsdbread in the stream scanner
-  TASK_STATUS__HALT,   // stream task will handle all data in the input queue, and then paused
+  TASK_STATUS__SCAN_HISTORY,  // stream task scan history data by using tsdbread in the stream scanner
+  TASK_STATUS__HALT,          // stream task will handle all data in the input queue, and then paused
   TASK_STATUS__PAUSE,
 };
 
@@ -266,11 +266,11 @@ typedef struct SCheckpointInfo {
 } SCheckpointInfo;
 
 typedef struct SStreamStatus {
-  int8_t taskStatus;
-  int8_t checkDownstream;
-  int8_t schedStatus;
-  int8_t keepTaskStatus;
-  bool   transferState;
+  int8_t        taskStatus;
+  int8_t        checkDownstream;
+  int8_t        schedStatus;
+  int8_t        keepTaskStatus;
+  bool          transferState;
   TdThreadMutex lock;
 } SStreamStatus;
 
@@ -280,19 +280,19 @@ typedef struct SHistDataRange {
 } SHistDataRange;
 
 typedef struct SSTaskBasicInfo {
-  int32_t         nodeId;       // vgroup id or snode id
-  SEpSet          epSet;
-  int32_t         selfChildId;
-  int32_t         totalLevel;
-  int8_t          taskLevel;
-  int8_t          fillHistory;  // is fill history task or not
+  int32_t nodeId;  // vgroup id or snode id
+  SEpSet  epSet;
+  int32_t selfChildId;
+  int32_t totalLevel;
+  int8_t  taskLevel;
+  int8_t  fillHistory;  // is fill history task or not
 } SSTaskBasicInfo;
 
 typedef struct SDispatchMsgInfo {
-  void*   pData;           // current dispatch data
-  int16_t msgType;         // dispatch msg type
-  int32_t retryCount;      // retry send data count
-  int64_t blockingTs;      // output blocking timestamp
+  void*   pData;       // current dispatch data
+  int16_t msgType;     // dispatch msg type
+  int32_t retryCount;  // retry send data count
+  int64_t blockingTs;  // output blocking timestamp
 } SDispatchMsgInfo;
 
 typedef struct {
@@ -351,21 +351,22 @@ struct SStreamTask {
 
 // meta
 typedef struct SStreamMeta {
-  char*        path;
-  TDB*         db;
-  TTB*         pTaskDb;
-  TTB*         pCheckpointDb;
-  SHashObj*    pTasks;
-  SArray*      pTaskList;  // SArray<task_id*>
-  void*        ahandle;
-  TXN*         txn;
-  FTaskExpand* expandFunc;
-  int32_t      vgId;
-  SRWLatch     lock;
-  int32_t      walScanCounter;
-  void*        streamBackend;
-  int64_t      streamBackendRid;
-  SHashObj*    pTaskBackendUnique;
+  char*         path;
+  TDB*          db;
+  TTB*          pTaskDb;
+  TTB*          pCheckpointDb;
+  SHashObj*     pTasks;
+  SArray*       pTaskList;  // SArray<task_id*>
+  void*         ahandle;
+  TXN*          txn;
+  FTaskExpand*  expandFunc;
+  int32_t       vgId;
+  SRWLatch      lock;
+  int32_t       walScanCounter;
+  void*         streamBackend;
+  int64_t       streamBackendRid;
+  SHashObj*     pTaskBackendUnique;
+  TdThreadMutex backendMutex;
 } SStreamMeta;
 
 int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
@@ -581,16 +582,16 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
 int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated);
 
 // common
-int32_t streamSetParamForScanHistoryData(SStreamTask* pTask);
-int32_t streamRestoreParam(SStreamTask* pTask);
-int32_t streamSetStatusNormal(SStreamTask* pTask);
+int32_t     streamSetParamForScanHistoryData(SStreamTask* pTask);
+int32_t     streamRestoreParam(SStreamTask* pTask);
+int32_t     streamSetStatusNormal(SStreamTask* pTask);
 const char* streamGetTaskStatusStr(int32_t status);
 
 // source level
-int32_t streamSetParamForStreamScanner(SStreamTask* pTask, SVersionRange *pVerRange, STimeWindow* pWindow);
+int32_t streamSetParamForStreamScanner(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
 int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated);
 int32_t streamSourceScanHistoryData(SStreamTask* pTask);
-//int32_t streamSourceRecoverScanStep2(SStreamTask* pTask, int64_t ver);
+// int32_t streamSourceRecoverScanStep2(SStreamTask* pTask, int64_t ver);
 int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask);
 
 int32_t streamDispatchTransferStateMsg(SStreamTask* pTask);

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

@@ -25,8 +25,9 @@
 #define SINK_NODE_LEVEL (0)
 extern bool tsDeployOnSnode;
 
-static int32_t mndAddSinkTaskToStream(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId, SVgObj* pVgroup, int32_t fillHistory);
-static void setFixedDownstreamEpInfo(SStreamTask* pDstTask, const SStreamTask* pTask);
+static int32_t mndAddSinkTaskToStream(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId,
+                                      SVgObj* pVgroup, int32_t fillHistory);
+static void    setFixedDownstreamEpInfo(SStreamTask* pDstTask, const SStreamTask* pTask);
 
 int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType,
                            int64_t watermark, int64_t deleteMark) {
@@ -101,13 +102,13 @@ int32_t mndSetSinkTaskInfo(SStreamObj* pStream, SStreamTask* pTask) {
   return 0;
 }
 
-int32_t mndAddDispatcherForInternalTask(SMnode* pMnode, SStreamObj* pStream, SArray* pSinkNodeList, SStreamTask* pTask) {
+int32_t mndAddDispatcherForInternalTask(SMnode* pMnode, SStreamObj* pStream, SArray* pSinkNodeList,
+                                        SStreamTask* pTask) {
   bool isShuffle = false;
 
   if (pStream->fixedSinkVgId == 0) {
     SDbObj* pDb = mndAcquireDb(pMnode, pStream->targetDb);
     if (pDb != NULL && pDb->cfg.numOfVgroups > 1) {
-
       isShuffle = true;
       pTask->outputType = TASK_OUTPUT__SHUFFLE_DISPATCH;
       pTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;
@@ -204,8 +205,8 @@ SVgObj* mndSchedFetchOneVg(SMnode* pMnode, int64_t dbUid) {
 
 // create sink node for each vgroup.
 int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SArray* pTaskList, SStreamObj* pStream, int32_t fillHistory) {
-  SSdb*   pSdb = pMnode->pSdb;
-  void*   pIter = NULL;
+  SSdb* pSdb = pMnode->pSdb;
+  void* pIter = NULL;
 
   while (1) {
     SVgObj* pVgroup = NULL;
@@ -226,7 +227,8 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SArray* pTaskList, SStrea
   return 0;
 }
 
-int32_t mndAddSinkTaskToStream(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId, SVgObj* pVgroup, int32_t fillHistory) {
+int32_t mndAddSinkTaskToStream(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId, SVgObj* pVgroup,
+                               int32_t fillHistory) {
   SStreamTask* pTask = tNewStreamTask(pStream->uid, TASK_LEVEL__SINK, fillHistory, 0, pTaskList);
   if (pTask == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -249,8 +251,8 @@ static int32_t addSourceStreamTask(SMnode* pMnode, SVgObj* pVgroup, SArray* pTas
 
   // todo set the correct ts, which should be last key of queried table.
   pTask->dataRange.window.skey = INT64_MIN;
-  pTask->dataRange.window.ekey = 1685959190000;//taosGetTimestampMs();
-//  pTask->dataRange.window.ekey = firstWindowSkey - 1;//taosGetTimestampMs();
+  pTask->dataRange.window.ekey = 1685959190000;  // taosGetTimestampMs();
+  //  pTask->dataRange.window.ekey = firstWindowSkey - 1;//taosGetTimestampMs();
 
   mDebug("add source task 0x%x window:%" PRId64 " - %" PRId64, pTask->id.taskId, pTask->dataRange.window.skey,
          pTask->dataRange.window.ekey);
@@ -300,10 +302,10 @@ int32_t setEpToDownstreamTask(SStreamTask* pTask, SStreamTask* pDownstream) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  if(pDownstream->pUpstreamEpInfoList == NULL) {
+  if (pDownstream->pUpstreamEpInfoList == NULL) {
     pDownstream->pUpstreamEpInfoList = taosArrayInit(4, POINTER_BYTES);
   }
-  
+
   taosArrayPush(pDownstream->pUpstreamEpInfoList, &pEpInfo);
   return TSDB_CODE_SUCCESS;
 }
@@ -316,7 +318,7 @@ static SArray* addNewTaskList(SArray* pTasksList) {
 
 // set the history task id
 static void setHTasksId(SArray* pTaskList, const SArray* pHTaskList) {
-  for(int32_t i = 0; i < taosArrayGetSize(pTaskList); ++i) {
+  for (int32_t i = 0; i < taosArrayGetSize(pTaskList); ++i) {
     SStreamTask** pStreamTask = taosArrayGet(pTaskList, i);
     SStreamTask** pHTask = taosArrayGet(pHTaskList, i);
 
@@ -341,7 +343,7 @@ static int32_t addSourceTasksForOneLevelStream(SMnode* pMnode, const SQueryPlan*
     pHTaskList = addNewTaskList(pStream->pHTasksList);
   }
 
-  SSdb* pSdb = pMnode->pSdb;
+  SSdb*          pSdb = pMnode->pSdb;
   SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
   if (LIST_LENGTH(inner->pNodeList) != 1) {
     terrno = TSDB_CODE_QRY_INVALID_INPUT;
@@ -369,8 +371,8 @@ static int32_t addSourceTasksForOneLevelStream(SMnode* pMnode, const SQueryPlan*
 
     // new stream task
     SArray** pSinkTaskList = taosArrayGet(pStream->tasks, SINK_NODE_LEVEL);
-    int32_t code = addSourceStreamTask(pMnode, pVgroup, pTaskList, *pSinkTaskList, pStream, plan, pStream->uid,
-                                       0, hasExtraSink, lastTs);
+    int32_t  code = addSourceStreamTask(pMnode, pVgroup, pTaskList, *pSinkTaskList, pStream, plan, pStream->uid, 0,
+                                        hasExtraSink, lastTs);
     if (code != TSDB_CODE_SUCCESS) {
       sdbRelease(pSdb, pVgroup);
       return -1;
@@ -392,8 +394,8 @@ static int32_t addSourceTasksForOneLevelStream(SMnode* pMnode, const SQueryPlan*
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t uid, SStreamTask* pDownstreamTask, SMnode* pMnode,
-                               SSubplan* pPlan, SVgObj* pVgroup) {
+static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t uid, SStreamTask* pDownstreamTask,
+                               SMnode* pMnode, SSubplan* pPlan, SVgObj* pVgroup) {
   SStreamTask* pTask = tNewStreamTask(uid, TASK_LEVEL__SOURCE, fillHistory, 0, pTaskList);
   if (pTask == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -402,7 +404,7 @@ static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t ui
 
   // todo set the correct ts, which should be last key of queried table.
   pTask->dataRange.window.skey = INT64_MIN;
-  pTask->dataRange.window.ekey = 1685959190000;//taosGetTimestampMs();
+  pTask->dataRange.window.ekey = 1685959190000;  // taosGetTimestampMs();
 
   mDebug("s-task:0x%x level:%d set time window:%" PRId64 " - %" PRId64, pTask->id.taskId, pTask->info.taskLevel,
          pTask->dataRange.window.skey, pTask->dataRange.window.ekey);
@@ -416,8 +418,8 @@ static int32_t doAddSourceTask(SArray* pTaskList, int8_t fillHistory, int64_t ui
   return setEpToDownstreamTask(pTask, pDownstreamTask);
 }
 
-static int32_t doAddAggTask(uint64_t uid, SArray* pTaskList, SArray* pSinkNodeList, SMnode* pMnode, SStreamObj* pStream, int32_t fillHistory,
-                            SStreamTask** pAggTask) {
+static int32_t doAddAggTask(uint64_t uid, SArray* pTaskList, SArray* pSinkNodeList, SMnode* pMnode, SStreamObj* pStream,
+                            int32_t fillHistory, SStreamTask** pAggTask) {
   *pAggTask = tNewStreamTask(uid, TASK_LEVEL__AGG, fillHistory, pStream->conf.triggerParam, pTaskList);
   if (*pAggTask == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -475,7 +477,8 @@ static int32_t addAggTask(SStreamObj* pStream, SMnode* pMnode, SQueryPlan* pPlan
     SArray* pHSinkNodeList = taosArrayGetP(pStream->pHTasksList, SINK_NODE_LEVEL);
 
     *pHAggTask = NULL;
-    code = doAddAggTask(pStream->hTaskUid, pHAggTaskList, pHSinkNodeList, pMnode, pStream, pStream->conf.fillHistory, pHAggTask);
+    code = doAddAggTask(pStream->hTaskUid, pHAggTaskList, pHSinkNodeList, pMnode, pStream, pStream->conf.fillHistory,
+                        pHAggTask);
     if (code != TSDB_CODE_SUCCESS) {
       if (pSnode != NULL) {
         sdbRelease(pSdb, pSnode);
@@ -541,7 +544,8 @@ static int32_t addSourceTasksForMultiLevelStream(SMnode* pMnode, SQueryPlan* pPl
     }
 
     if (pStream->conf.fillHistory) {
-      code = doAddSourceTask(pHSourceTaskList, pStream->conf.fillHistory, pStream->hTaskUid, pHDownstreamTask, pMnode, plan, pVgroup);
+      code = doAddSourceTask(pHSourceTaskList, pStream->conf.fillHistory, pStream->hTaskUid, pHDownstreamTask, pMnode,
+                             plan, pVgroup);
       sdbRelease(pSdb, pVgroup);
 
       if (code != TSDB_CODE_SUCCESS) {
@@ -555,7 +559,8 @@ static int32_t addSourceTasksForMultiLevelStream(SMnode* pMnode, SQueryPlan* pPl
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t addSinkTasks(SArray* pTasksList, SMnode* pMnode, SStreamObj* pStream, SArray** pCreatedTaskList, int32_t fillHistory) {
+static int32_t addSinkTasks(SArray* pTasksList, SMnode* pMnode, SStreamObj* pStream, SArray** pCreatedTaskList,
+                            int32_t fillHistory) {
   SArray* pSinkTaskList = addNewTaskList(pTasksList);
   if (pStream->fixedSinkVgId == 0) {
     if (mndAddShuffleSinkTasksToStream(pMnode, pSinkTaskList, pStream, fillHistory) < 0) {
@@ -563,7 +568,8 @@ static int32_t addSinkTasks(SArray* pTasksList, SMnode* pMnode, SStreamObj* pStr
       return -1;
     }
   } else {
-    if (mndAddSinkTaskToStream(pStream, pSinkTaskList, pMnode, pStream->fixedSinkVgId, &pStream->fixedSinkVg, fillHistory) < 0) {
+    if (mndAddSinkTaskToStream(pStream, pSinkTaskList, pMnode, pStream->fixedSinkVgId, &pStream->fixedSinkVg,
+                               fillHistory) < 0) {
       // TODO free
       return -1;
     }
@@ -574,7 +580,7 @@ static int32_t addSinkTasks(SArray* pTasksList, SMnode* pMnode, SStreamObj* pStr
 }
 
 static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan* pPlan, int64_t lastTs) {
-  SSdb* pSdb = pMnode->pSdb;
+  SSdb*   pSdb = pMnode->pSdb;
   int32_t numOfPlanLevel = LIST_LENGTH(pPlan->pSubplans);
 
   bool    hasExtraSink = false;
@@ -658,8 +664,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
       terrno = TSDB_CODE_QRY_INVALID_INPUT;
       return -1;
     }
-  }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE && pTopic->ast != NULL){
-    SNode *pAst = NULL;
+  } else if (pTopic->subType == TOPIC_SUB_TYPE__TABLE && pTopic->ast != NULL) {
+    SNode* pAst = NULL;
     if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
       mError("topic:%s, failed to create since %s", pTopic->name, terrstr());
       return -1;
@@ -674,7 +680,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     nodesDestroyNode(pAst);
   }
 
-  if(pPlan){
+  if (pPlan) {
     int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
     if (levelNum != 1) {
       qDestroyQueryPlan(pPlan);

source/libs/stream/inc/streamBackendRocksdb.h

@@ -42,10 +42,11 @@ typedef struct {
   TdThreadMutex                      cfMutex;
   SHashObj*                          cfInst;
   int64_t                            defaultCfInit;
-} SBackendHandle;
+} SBackendWrapper;
 
 void*      streamBackendInit(const char* path);
 void       streamBackendCleanup(void* arg);
+void       streamBackendHandleCleanup(void* arg);
 SListNode* streamBackendAddCompare(void* backend, void* arg);
 void       streamBackendDelCompare(void* backend, void* arg);
 

source/libs/stream/inc/streamInc.h

@@ -50,11 +50,12 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
 int32_t streamDispatchCheckMsg(SStreamTask* pTask, const SStreamTaskCheckReq* pReq, int32_t nodeId, SEpSet* pEpSet);
 
 int32_t streamDoDispatchScanHistoryFinishMsg(SStreamTask* pTask, const SStreamRecoverFinishReq* pReq, int32_t vgId,
-                                          SEpSet* pEpSet);
+                                             SEpSet* pEpSet);
 
 SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem);
 
 extern int32_t streamBackendId;
+extern int32_t streamBackendCfWrapperId;
 
 #ifdef __cplusplus
 }

source/libs/stream/src/streamMeta.c

@@ -21,10 +21,18 @@
 
 static TdThreadOnce streamMetaModuleInit = PTHREAD_ONCE_INIT;
 int32_t             streamBackendId = 0;
-static void         streamMetaEnvInit() { streamBackendId = taosOpenRef(20, streamBackendCleanup); }
+int32_t             streamBackendCfWrapperId = 0;
+
+static void streamMetaEnvInit() {
+  streamBackendId = taosOpenRef(64, streamBackendCleanup);
+  streamBackendCfWrapperId = taosOpenRef(64, streamBackendHandleCleanup);
+}
 
 void streamMetaInit() { taosThreadOnce(&streamMetaModuleInit, streamMetaEnvInit); }
-void streamMetaCleanup() { taosCloseRef(streamBackendId); }
+void streamMetaCleanup() {
+  taosCloseRef(streamBackendId);
+  taosCloseRef(streamBackendCfWrapperId);
+}
 
 SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId) {
   int32_t      code = -1;
@@ -93,10 +101,14 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
     goto _err;
   }
   pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
+  pMeta->pTaskBackendUnique =
+      taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
 
   taosMemoryFree(streamPath);
 
   taosInitRWLatch(&pMeta->lock);
+  taosThreadMutexInit(&pMeta->backendMutex, NULL);
+
   return pMeta;
 
 _err:
@@ -139,6 +151,8 @@ void streamMetaClose(SStreamMeta* pMeta) {
   taosRemoveRef(streamBackendId, pMeta->streamBackendRid);
   pMeta->pTaskList = taosArrayDestroy(pMeta->pTaskList);
   taosMemoryFree(pMeta->path);
+  taosThreadMutexDestroy(&pMeta->backendMutex);
+  taosHashCleanup(pMeta->pTaskBackendUnique);
   taosMemoryFree(pMeta);
 }
 

source/libs/stream/src/streamState.c

@@ -116,16 +116,33 @@ SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t sz
 
   pState->taskId = pStreamTask->id.taskId;
   pState->streamId = pStreamTask->id.streamId;
+  sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-%d", pState->streamId, pState->taskId);
 
 #ifdef USE_ROCKSDB
   SStreamMeta* pMeta = pStreamTask->pMeta;
   pState->streamBackendRid = pMeta->streamBackendRid;
-  int code = streamStateOpenBackend(pMeta->streamBackend, pState);
-  if (code == -1) {
-    taosReleaseRef(streamBackendId, pMeta->streamBackendRid);
-    taosMemoryFree(pState);
-    pState = NULL;
+  // taosWLockLatch(&pMeta->lock);
+  taosThreadMutexLock(&pMeta->backendMutex);
+  void* uniqueId =
+      taosHashGet(pMeta->pTaskBackendUnique, pState->pTdbState->idstr, strlen(pState->pTdbState->idstr) + 1);
+  if (uniqueId == NULL) {
+    int code = streamStateOpenBackend(pMeta->streamBackend, pState);
+    if (code == -1) {
+      taosReleaseRef(streamBackendId, pState->streamBackendRid);
+      taosThreadMutexUnlock(&pMeta->backendMutex);
+      taosMemoryFree(pState);
+      return NULL;
+    }
+    taosHashPut(pMeta->pTaskBackendUnique, pState->pTdbState->idstr, strlen(pState->pTdbState->idstr) + 1,
+                &pState->pTdbState->backendCfWrapperId, sizeof(pState->pTdbState->backendCfWrapperId));
+  } else {
+    int64_t id = *(int64_t*)uniqueId;
+    pState->pTdbState->backendCfWrapperId = id;
+    pState->pTdbState->pBackendCfWrapper = taosAcquireRef(streamBackendCfWrapperId, id);
+
+    taosAcquireRef(streamBackendId, pState->streamBackendRid);
   }
+  taosThreadMutexUnlock(&pMeta->backendMutex);
 
   pState->pTdbState->pOwner = pTask;
   pState->pFileState = NULL;
@@ -385,8 +402,8 @@ int32_t streamStateClear(SStreamState* pState) {
   streamStatePut(pState, &key, NULL, 0);
   while (1) {
     SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &key);
-    SWinKey delKey = {0};
-    int32_t code = streamStateGetKVByCur(pCur, &delKey, NULL, 0);
+    SWinKey          delKey = {0};
+    int32_t          code = streamStateGetKVByCur(pCur, &delKey, NULL, 0);
     streamStateFreeCur(pCur);
     if (code == 0) {
       streamStateDel(pState, &delKey);
@@ -498,7 +515,7 @@ int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void**
     return -1;
   }
   const SStateKey* pKTmp = NULL;
-  int32_t kLen;
+  int32_t          kLen;
   if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
     return -1;
   }
@@ -518,7 +535,7 @@ int32_t streamStateFillGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const vo
     return -1;
   }
   const SWinKey* pKTmp = NULL;
-  int32_t kLen;
+  int32_t        kLen;
   if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
     return -1;
   }
@@ -535,7 +552,7 @@ int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const v
     return -1;
   }
   uint64_t groupId = pKey->groupId;
-  int32_t code = streamStateFillGetKVByCur(pCur, pKey, pVal, pVLen);
+  int32_t  code = streamStateFillGetKVByCur(pCur, pKey, pVal, pVLen);
   if (code == 0) {
     if (pKey->groupId == groupId) {
       return 0;
@@ -553,7 +570,7 @@ int32_t streamStateGetFirst(SStreamState* pState, SWinKey* key) {
   SWinKey tmp = {.ts = 0, .groupId = 0};
   streamStatePut(pState, &tmp, NULL, 0);
   SStreamStateCur* pCur = streamStateSeekKeyNext(pState, &tmp);
-  int32_t code = streamStateGetKVByCur(pCur, key, NULL, 0);
+  int32_t          code = streamStateGetKVByCur(pCur, key, NULL, 0);
   streamStateFreeCur(pCur);
   streamStateDel(pState, &tmp);
   return code;
@@ -593,7 +610,7 @@ SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key
   }
 
   SStateKey sKey = {.key = *key, .opNum = pState->number};
-  int32_t c = 0;
+  int32_t   c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c) < 0) {
     streamStateFreeCur(pCur);
     return NULL;
@@ -726,9 +743,9 @@ int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVa
 #else
 
   SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pState, key);
-  SSessionKey resKey = *key;
-  void* tmp = NULL;
-  int32_t code = streamStateSessionGetKVByCur(pCur, &resKey, &tmp, pVLen);
+  SSessionKey      resKey = *key;
+  void*            tmp = NULL;
+  int32_t          code = streamStateSessionGetKVByCur(pCur, &resKey, &tmp, pVLen);
   if (code == 0) {
     if (key->win.skey != resKey.win.skey) {
       code = -1;
@@ -767,7 +784,7 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, cons
   }
 
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
-  int32_t c = 0;
+  int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
     streamStateFreeCur(pCur);
     return NULL;
@@ -798,7 +815,7 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentNext(SStreamState* pState, cons
   }
 
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
-  int32_t c = 0;
+  int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
     streamStateFreeCur(pCur);
     return NULL;
@@ -830,7 +847,7 @@ SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSess
   }
 
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
-  int32_t c = 0;
+  int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
     streamStateFreeCur(pCur);
     return NULL;
@@ -854,7 +871,7 @@ int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, v
     return -1;
   }
   SStateSessionKey* pKTmp = NULL;
-  int32_t kLen;
+  int32_t           kLen;
   if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, (const void**)pVal, pVLen) < 0) {
     return -1;
   }
@@ -873,13 +890,13 @@ int32_t streamStateSessionClear(SStreamState* pState) {
 #ifdef USE_ROCKSDB
   return streamStateSessionClear_rocksdb(pState);
 #else
-  SSessionKey key = {.win.skey = 0, .win.ekey = 0, .groupId = 0};
+  SSessionKey      key = {.win.skey = 0, .win.ekey = 0, .groupId = 0};
   SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pState, &key);
   while (1) {
     SSessionKey delKey = {0};
-    void* buf = NULL;
-    int32_t size = 0;
-    int32_t code = streamStateSessionGetKVByCur(pCur, &delKey, &buf, &size);
+    void*       buf = NULL;
+    int32_t     size = 0;
+    int32_t     code = streamStateSessionGetKVByCur(pCur, &delKey, &buf, &size);
     if (code == 0 && size > 0) {
       memset(buf, 0, size);
       streamStateSessionPut(pState, &delKey, buf, size);
@@ -908,14 +925,14 @@ int32_t streamStateSessionGetKeyByRange(SStreamState* pState, const SSessionKey*
   }
 
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
-  int32_t c = 0;
+  int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
     streamStateFreeCur(pCur);
     return -1;
   }
 
   SSessionKey resKey = *key;
-  int32_t code = streamStateSessionGetKVByCur(pCur, &resKey, NULL, 0);
+  int32_t     code = streamStateSessionGetKVByCur(pCur, &resKey, NULL, 0);
   if (code == 0 && sessionRangeKeyCmpr(key, &resKey) == 0) {
     *curKey = resKey;
     streamStateFreeCur(pCur);
@@ -951,19 +968,19 @@ int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key,
   return streamStateSessionAddIfNotExist_rocksdb(pState, key, gap, pVal, pVLen);
 #else
   // todo refactor
-  int32_t res = 0;
+  int32_t     res = 0;
   SSessionKey originKey = *key;
   SSessionKey searchKey = *key;
   searchKey.win.skey = key->win.skey - gap;
   searchKey.win.ekey = key->win.ekey + gap;
   int32_t valSize = *pVLen;
-  void* tmp = tdbRealloc(NULL, valSize);
+  void*   tmp = tdbRealloc(NULL, valSize);
   if (!tmp) {
     return -1;
   }
 
   SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentPrev(pState, key);
-  int32_t code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
+  int32_t          code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
   if (code == 0) {
     if (sessionRangeKeyCmpr(&searchKey, key) == 0) {
       memcpy(tmp, *pVal, valSize);
@@ -1006,16 +1023,16 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
 #ifdef USE_ROCKSDB
   return streamStateStateAddIfNotExist_rocksdb(pState, key, pKeyData, keyDataLen, fn, pVal, pVLen);
 #else
-  int32_t res = 0;
+  int32_t     res = 0;
   SSessionKey tmpKey = *key;
-  int32_t valSize = *pVLen;
-  void* tmp = tdbRealloc(NULL, valSize);
+  int32_t     valSize = *pVLen;
+  void*       tmp = tdbRealloc(NULL, valSize);
   if (!tmp) {
     return -1;
   }
 
   SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentPrev(pState, key);
-  int32_t code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
+  int32_t          code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
   if (code == 0) {
     if (key->win.skey <= tmpKey.win.skey && tmpKey.win.ekey <= key->win.ekey) {
       memcpy(tmp, *pVal, valSize);
@@ -1113,9 +1130,7 @@ int32_t streamStateDeleteCheckPoint(SStreamState* pState, TSKEY mark) {
 #endif
 }
 
-void streamStateReloadInfo(SStreamState* pState, TSKEY ts) {
-  streamFileStateReloadInfo(pState->pFileState, ts);
-}
+void streamStateReloadInfo(SStreamState* pState, TSKEY ts) { streamFileStateReloadInfo(pState->pFileState, ts); }
 
 #if 0
 char* streamStateSessionDump(SStreamState* pState) {