taosdata · hjxilinx · Jul 5, 2023 · May 30, 2023 · May 30, 2023 · May 30, 2023
diff --git a/cmake/rocksdb_CMakeLists.txt.in b/cmake/rocksdb_CMakeLists.txt.in
@@ -5,8 +5,8 @@ if (${BUILD_CONTRIB})
 	        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" 
+	        DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
+	        SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb"
             CONFIGURE_COMMAND ""
             BUILD_COMMAND ""
             INSTALL_COMMAND ""
@@ -18,8 +18,8 @@ else()
 	        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" 
+	        DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
+	        SOURCE_DIR "${TD_CONTRIB_DIR}/rocksdb"
             CONFIGURE_COMMAND ""
             BUILD_COMMAND ""
             INSTALL_COMMAND ""

diff --git a/include/common/tcommon.h b/include/common/tcommon.h
@@ -54,6 +54,11 @@ typedef struct SSessionKey {
   uint64_t    groupId;
 } SSessionKey;
 
+typedef struct SVersionRange {
+  uint64_t minVer;
+  uint64_t maxVer;
+} SVersionRange;
+
 static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
   SWinKey* pWin1 = (SWinKey*)pKey1;
   SWinKey* pWin2 = (SWinKey*)pKey2;

diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h
@@ -177,7 +177,6 @@ static FORCE_INLINE void colDataSetDouble(SColumnInfoData* pColumnInfoData, uint
 int32_t getJsonValueLen(const char* data);
 
 int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
-int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
 int32_t colDataReassignVal(SColumnInfoData* pColumnInfoData, uint32_t dstRowIdx, uint32_t srcRowIdx, const char* pData);
 int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows, bool trimValue);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
@@ -187,6 +186,7 @@ int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* p
 int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex);
 
 int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows);
+
 int32_t colDataGetRowLength(const SColumnInfoData* pColumnInfoData, int32_t rowIdx);
 void    colDataTrim(SColumnInfoData* pColumnInfoData);
 
@@ -208,7 +208,6 @@ double blockDataGetSerialRowSize(const SSDataBlock* pBlock);
 size_t blockDataGetSerialMetaSize(uint32_t numOfCols);
 
 int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo);
-int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
 
 int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload);
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
@@ -237,11 +236,10 @@ int32_t      blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColIn
 SColumnInfoData  createColumnInfoData(int16_t type, int32_t bytes, int16_t colId);
 SColumnInfoData* bdGetColumnInfoData(const SSDataBlock* pBlock, int32_t index);
 
+int32_t blockGetEncodeSize(const SSDataBlock* pBlock);
 int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols);
 const char* blockDecode(SSDataBlock* pBlock, const char* pData);
 
-void blockDebugShowDataBlock(SSDataBlock* pBlock, const char* flag);
-void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag);
 // for debug
 char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
@@ -251,9 +249,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pData
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
 int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);
 
-static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) {
-  return blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock)) + blockDataGetSize(pBlock);
-}
+void trimDataBlock(SSDataBlock* pBlock, int32_t totalRows, const bool* pBoolList);
 
 #ifdef __cplusplus
 }

diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h
@@ -252,7 +252,9 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DISPATCH, "stream-task-dispatch", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_UNUSED1, "stream-unused1", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE, "stream-retrieve", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_STREAM_RECOVER_FINISH, "stream-recover-finish", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_SCAN_HISTORY, "stream-scan-history", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_SCAN_HISTORY_FINISH, "stream-scan-history-finish", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_TRANSFER_STATE, "stream-transfer-state", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_CHECK, "stream-task-check", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_CHECKPOINT, "stream-checkpoint", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_REPORT_CHECKPOINT, "stream-report-checkpoint", NULL, NULL)
@@ -297,8 +299,7 @@ enum {
 
   TD_NEW_MSG_SEG(TDMT_VND_STREAM_MSG)
   TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TRIGGER, "vnode-stream-trigger", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_NONBLOCKING_STAGE, "vnode-stream-recover1", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE, "vnode-stream-recover2", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_SCAN_HISTORY, "vnode-stream-scan-history", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_STREAM_CHECK_POINT_SOURCE, "vnode-stream-checkpoint-source", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_STREAM_MAX_MSG, "vnd-stream-max", NULL, NULL)
 

diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h
@@ -55,6 +55,9 @@ typedef struct {
 
   void*   pStateBackend;
   struct SStorageAPI api;
+
+  int8_t        fillHistory;
+  STimeWindow   winRange;
 } SReadHandle;
 
 // in queue mode, data streams are seperated by msg
@@ -193,14 +196,6 @@ int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t le
 void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order);
 void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery);
 STimeWindow getAlignQueryTimeWindow(const SInterval* pInterval, int64_t key);
-/**
- * return the scan info, in the form of tuple of two items, including table uid and current timestamp
- * @param tinfo
- * @param uid
- * @param ts
- * @return
- */
-int32_t qGetStreamScanStatus(qTaskInfo_t tinfo, uint64_t* uid, int64_t* ts);
 
 SArray* qGetQueriedTableListInfo(qTaskInfo_t tinfo);
 
@@ -220,15 +215,22 @@ void* qExtractReaderFromStreamScanner(void* scanner);
 
 int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner);
 
-int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo);
-int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver);
-int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver);
+int32_t qSetStreamOperatorOptionForScanHistory(qTaskInfo_t tinfo);
+int32_t qStreamSourceScanParamForHistoryScanStep1(qTaskInfo_t tinfo, SVersionRange *pVerRange, STimeWindow* pWindow);
+int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRange *pVerRange, STimeWindow* pWindow);
 int32_t qStreamRecoverFinish(qTaskInfo_t tinfo);
-int32_t qStreamRestoreParam(qTaskInfo_t tinfo);
+int32_t qRestoreStreamOperatorOption(qTaskInfo_t tinfo);
 bool    qStreamRecoverScanFinished(qTaskInfo_t tinfo);
-void    qStreamCloseTsdbReader(void* task);
+bool    qStreamRecoverScanStep1Finished(qTaskInfo_t tinfo);
+bool    qStreamRecoverScanStep2Finished(qTaskInfo_t tinfo);
+int32_t qStreamRecoverSetAllStepFinished(qTaskInfo_t tinfo);
 void    resetTaskInfo(qTaskInfo_t tinfo);
 
+void qResetStreamInfoTimeWindow(qTaskInfo_t tinfo);
+
+int32_t qStreamOperatorReleaseState(qTaskInfo_t tInfo);
+int32_t qStreamOperatorReloadState(qTaskInfo_t tInfo);
+
 #ifdef __cplusplus
 }
 #endif

diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h
@@ -234,29 +234,6 @@ typedef struct SStoreSnapshotFn {
   int32_t (*getTableInfoFromSnapshot)(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid);
 } SStoreSnapshotFn;
 
-/**
-void        metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
-void        metaReaderReleaseLock(SMetaReader *pReader);
-void        metaReaderClear(SMetaReader *pReader);
-int32_t     metaReaderGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
-int32_t     metaReaderGetTableEntryByUidCache(SMetaReader *pReader, tb_uid_t uid);
-int32_t     metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList);
-const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal);
-int         metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
-
-int      metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid);
-int      metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType);
-bool     metaIsTableExist(SMeta *pMeta, tb_uid_t uid);
-int32_t  metaGetCachedTableUidList(SMeta *pMeta, tb_uid_t suid, const uint8_t *key, int32_t keyLen, SArray *pList,
-                                   bool *acquired);
-int32_t  metaUidFilterCachePut(SMeta *pMeta, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
-                               int32_t payloadLen, double selectivityRatio);
-tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name);
-int32_t  metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList);
-int32_t  metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t
-payloadLen);
- */
-
 typedef struct SStoreMeta {
   SMTbCursor* (*openTableMetaCursor)(void* pVnode);                     // metaOpenTbCursor
   void (*closeTableMetaCursor)(SMTbCursor* pTbCur);                     // metaCloseTbCursor
@@ -403,7 +380,7 @@ typedef struct SStateStore {
   SStreamStateCur* (*streamStateSessionSeekKeyCurrentNext)(SStreamState* pState, const SSessionKey* key);
 
   struct SStreamFileState* (*streamFileStateInit)(int64_t memSize, uint32_t keySize, uint32_t rowSize,
-                                                  uint32_t selectRowSize, GetTsFun fp, void* pFile, TSKEY delMark);
+                                                  uint32_t selectRowSize, GetTsFun fp, void* pFile, TSKEY delMark, const char*id);
 
   void (*streamFileStateDestroy)(struct SStreamFileState* pFileState);
   void (*streamFileStateClear)(struct SStreamFileState* pFileState);
@@ -415,6 +392,7 @@ typedef struct SStateStore {
   int32_t (*streamStateCommit)(SStreamState* pState);
   void (*streamStateDestroy)(SStreamState* pState, bool remove);
   int32_t (*streamStateDeleteCheckPoint)(SStreamState* pState, TSKEY mark);
+  void (*streamStateReloadInfo)(SStreamState* pState, TSKEY ts);
 } SStateStore;
 
 typedef struct SStorageAPI {

diff --git a/include/libs/function/function.h b/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 {

diff --git a/include/libs/stream/streamState.h b/include/libs/stream/streamState.h
@@ -138,6 +138,8 @@ int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char* tbname);
 int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal);
 
+void streamStateReloadInfo(SStreamState* pState, TSKEY ts);
+
 /***compare func **/
 
 typedef struct SStateChekpoint {