diff --git a/include/common/tcommon.h b/include/common/tcommon.h
index d02371dc1a946ab16d885fe5a4154cf6f919bbe7..a97c68be4920333f6a08f02fdb6753724db035a3 100644
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -51,15 +51,15 @@ static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
SWinKey* pWin1 = (SWinKey*)pKey1;
SWinKey* pWin2 = (SWinKey*)pKey2;
- if (pWin1->ts > pWin2->ts) {
+ if (pWin1->groupId > pWin2->groupId) {
return 1;
- } else if (pWin1->ts < pWin2->ts) {
+ } else if (pWin1->groupId < pWin2->groupId) {
return -1;
}
- if (pWin1->groupId > pWin2->groupId) {
+ if (pWin1->ts > pWin2->ts) {
return 1;
- } else if (pWin1->groupId < pWin2->groupId) {
+ } else if (pWin1->ts < pWin2->ts) {
return -1;
}
diff --git a/include/libs/stream/streamState.h b/include/libs/stream/streamState.h
index 6103262e1775e94a780901f2764984a5e913b834..ae84299c1c02d6cc7c1f1b30ac07bbf621dc294f 100644
--- a/include/libs/stream/streamState.h
+++ b/include/libs/stream/streamState.h
@@ -38,6 +38,8 @@ typedef struct STdbState {
rocksdb_comparator_t** pCompare;
rocksdb_options_t* dbOpt;
struct SStreamTask* pOwner;
+ void* param;
+ void* env;
TDB* db;
TTB* pStateDb;
@@ -63,6 +65,7 @@ void streamStateClose(SStreamState* pState);
int32_t streamStateBegin(SStreamState* pState);
int32_t streamStateCommit(SStreamState* pState);
void streamStateDestroy(SStreamState* pState);
+int32_t streamStateDeleteCheckPoint(SStreamState* pState, TSKEY mark);
typedef struct {
rocksdb_iterator_t* iter;
diff --git a/include/libs/stream/tstreamFileState.h b/include/libs/stream/tstreamFileState.h
index 1962e41b6cb5e142cbeadfe51599812aae26aab0..25267be3548427597594906491c155e5f25afe48 100644
--- a/include/libs/stream/tstreamFileState.h
+++ b/include/libs/stream/tstreamFileState.h
@@ -18,6 +18,7 @@
#include "os.h"
+#include "tarray.h"
#include "tdef.h"
#include "tlist.h"
@@ -27,29 +28,33 @@ extern "C" {
typedef struct SStreamFileState SStreamFileState;
typedef struct SRowBuffPos {
- void* pRowBuff;
- void* pKey;
- bool beFlushed;
- bool beUsed;
+ void* pRowBuff;
+ void* pKey;
+ bool beFlushed;
+ bool beUsed;
} SRowBuffPos;
typedef SList SStreamSnapshot;
typedef TSKEY (*GetTsFun)(void*);
-SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, GetTsFun fp, void* pFile, TSKEY delMark);
+SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, GetTsFun fp, void* pFile,
+ TSKEY delMark);
void streamFileStateDestroy(SStreamFileState* pFileState);
void streamFileStateClear(SStreamFileState* pFileState);
int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen);
int32_t deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLen);
int32_t getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void** pVal);
-void releaseRowBuffPos(SRowBuffPos* pBuff);
-bool hasRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen);
+void releaseRowBuffPos(SRowBuffPos* pBuff);
+bool hasRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen);
SStreamSnapshot* getSnapshot(SStreamFileState* pFileState);
-int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState);
-int32_t recoverSnapshot(SStreamFileState* pFileState);
+int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState);
+int32_t recoverSnapshot(SStreamFileState* pFileState);
+
+int32_t getSnapshotIdList(SStreamFileState* pFileState, SArray* list);
+int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark);
#ifdef __cplusplus
}
diff --git a/include/libs/stream/tstreamUpdate.h b/include/libs/stream/tstreamUpdate.h
index ab328c6ad5c20170bf6b3c7ca7f4930b9082ad17..4678aa0bd9630beff4eb53111dbe5221325d1ac9 100644
--- a/include/libs/stream/tstreamUpdate.h
+++ b/include/libs/stream/tstreamUpdate.h
@@ -40,9 +40,7 @@ typedef struct SUpdateInfo {
TSKEY minTS;
SScalableBf *pCloseWinSBF;
SHashObj *pMap;
- STimeWindow scanWindow;
- uint64_t scanGroupId;
- uint64_t maxVersion;
+ uint64_t maxDataVersion;
} SUpdateInfo;
SUpdateInfo *updateInfoInitP(SInterval *pInterval, int64_t watermark);
@@ -50,8 +48,6 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma
TSKEY updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t primaryTsCol);
bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts);
bool updateInfoIsTableInserted(SUpdateInfo *pInfo, int64_t tbUid);
-void updateInfoSetScanRange(SUpdateInfo *pInfo, STimeWindow *pWin, uint64_t groupId, uint64_t version);
-bool updateInfoIgnore(SUpdateInfo *pInfo, STimeWindow *pWin, uint64_t groupId, uint64_t version);
void updateInfoDestroy(SUpdateInfo *pInfo);
void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo);
void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo);
diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index 5d08f678e884992942aa609c1bfab88ab08b6869..f2b62b117425cd3fcb01ade0bf9ea25e8df57fdb 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -477,9 +477,10 @@ typedef struct SStreamScanInfo {
int32_t blockRecoverTotCnt;
SSDataBlock* pRecoverRes;
- SSDataBlock* pCreateTbRes;
- int8_t igCheckUpdate;
- int8_t igExpired;
+ SSDataBlock* pCreateTbRes;
+ int8_t igCheckUpdate;
+ int8_t igExpired;
+ SStreamState* pState;
} SStreamScanInfo;
typedef struct {
@@ -553,6 +554,7 @@ typedef struct SStreamIntervalOperatorInfo {
SSDataBlock* pPullDataRes;
bool isFinal;
SArray* pChildren;
+ int32_t numOfChild;
SStreamState* pState;
SWinKey delKey;
uint64_t numOfDatapack;
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 0e5460fd8c2071f228c8faab96488448466da9a0..bc19572394418f67e35f23c4f721a889972f8171 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -1034,8 +1034,9 @@ static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t grou
pInfo->groupId = groupCol[rowIndex];
}
-void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
+void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin, uint64_t version) {
pTableScanInfo->base.cond.twindows = *pWin;
+ pTableScanInfo->base.cond.endVersion = version;
pTableScanInfo->scanTimes = 0;
pTableScanInfo->currentGroupId = -1;
tsdbReaderClose(pTableScanInfo->base.dataReader);
@@ -1154,7 +1155,7 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_
break;
}
- resetTableScanInfo(pInfo->pTableScanOp->info, &win);
+ resetTableScanInfo(pInfo->pTableScanOp->info, &win, pInfo->pUpdateInfo->maxDataVersion);
pInfo->pTableScanOp->status = OP_OPENED;
return true;
}
@@ -1195,14 +1196,19 @@ static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, uint64_t
static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
qInfo("do stream range scan. windows index:%d", *pRowIndex);
+ bool prepareRes = true;
while (1) {
SSDataBlock* pResult = NULL;
pResult = doTableScan(pInfo->pTableScanOp);
- if (!pResult && prepareRangeScan(pInfo, pSDB, pRowIndex)) {
+ if (!pResult) {
+ prepareRes = prepareRangeScan(pInfo, pSDB, pRowIndex);
// scan next window data
pResult = doTableScan(pInfo->pTableScanOp);
}
if (!pResult) {
+ if (prepareRes) {
+ continue;
+ }
blockDataCleanup(pSDB);
*pRowIndex = 0;
pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
@@ -1453,39 +1459,8 @@ static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock,
return code;
}
-#if 0
-void calBlockTag(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
- SExprSupp* pTagCalSup = &pInfo->tagCalSup;
- SStreamState* pState = pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState;
- if (pTagCalSup == NULL || pTagCalSup->numOfExprs == 0) return;
- if (pBlock == NULL || pBlock->info.rows == 0) return;
-
- void* tag = NULL;
- int32_t tagLen = 0;
- if (streamStateGetParTag(pState, pBlock->info.id.groupId, &tag, &tagLen) == 0) {
- pBlock->info.tagLen = tagLen;
- void* pTag = taosMemoryRealloc(pBlock->info.pTag, tagLen);
- if (pTag == NULL) {
- tdbFree(tag);
- taosMemoryFree(pBlock->info.pTag);
- pBlock->info.pTag = NULL;
- pBlock->info.tagLen = 0;
- return;
- }
- pBlock->info.pTag = pTag;
- memcpy(pBlock->info.pTag, tag, tagLen);
- tdbFree(tag);
- return;
- } else {
- pBlock->info.pTag = NULL;
- }
- tdbFree(tag);
-}
-#endif
-
static void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
SExprSupp* pTbNameCalSup = &pInfo->tbnameCalSup;
- SStreamState* pState = pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState;
blockDataCleanup(pInfo->pCreateTbRes);
if (pInfo->tbnameCalSup.numOfExprs == 0 && pInfo->tagCalSup.numOfExprs == 0) {
pBlock->info.parTbName[0] = 0;
@@ -1541,7 +1516,7 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.id.uid, tsCol[rowId]);
bool closedWin = isClosed && isSignleIntervalWindow(pInfo) &&
isDeletedStreamWindow(&win, pBlock->info.id.groupId,
- pInfo->pTableScanOp->pTaskInfo->streamInfo.pState, &pInfo->twAggSup);
+ pInfo->pState, &pInfo->twAggSup);
if ((update || closedWin) && out) {
qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
uint64_t gpId = 0;
@@ -1770,6 +1745,7 @@ static void setBlockGroupIdByUid(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
static void doCheckUpdate(SStreamScanInfo* pInfo, TSKEY endKey, SSDataBlock* pBlock) {
if (pInfo->pUpdateInfo) {
+ pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pBlock->info.version);
checkUpdateData(pInfo, true, pBlock, true);
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, endKey);
if (pInfo->pUpdateDataRes->info.rows > 0) {
@@ -1831,7 +1807,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN2;
}
- /*resetTableScanInfo(pTSInfo, pWin);*/
tsdbReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL;
@@ -1876,8 +1851,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
if (pSDB) {
STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
- uint64_t version = getReaderMaxVersion(pTableScanInfo->base.dataReader);
- updateInfoSetScanRange(pInfo->pUpdateInfo, &pTableScanInfo->base.cond.twindows, pInfo->groupId, version);
pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA;
checkUpdateData(pInfo, true, pSDB, false);
printDataBlock(pSDB, "scan recover update");
@@ -1946,6 +1919,9 @@ FETCH_NEXT_BLOCK:
pBlock->info.calWin.skey = INT64_MIN;
pBlock->info.calWin.ekey = INT64_MAX;
pBlock->info.dataLoad = 1;
+ if (pInfo->pUpdateInfo) {
+ pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pBlock->info.version);
+ }
blockDataUpdateTsWindow(pBlock, 0);
switch (pBlock->info.type) {
case STREAM_NORMAL:
@@ -2043,11 +2019,9 @@ FETCH_NEXT_BLOCK:
SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
if (pSDB) {
STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
- uint64_t version = getReaderMaxVersion(pTableScanInfo->base.dataReader);
- updateInfoSetScanRange(pInfo->pUpdateInfo, &pTableScanInfo->base.cond.twindows, pInfo->groupId, version);
pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA;
checkUpdateData(pInfo, true, pSDB, false);
- // printDataBlock(pSDB, "stream scan update");
+ printDataBlock(pSDB, "stream scan update");
calBlockTbName(pInfo, pSDB);
return pSDB;
}
@@ -2109,13 +2083,6 @@ FETCH_NEXT_BLOCK:
setBlockIntoRes(pInfo, &block, false);
- if (updateInfoIgnore(pInfo->pUpdateInfo, &pInfo->pRes->info.window, pInfo->pRes->info.id.groupId,
- pInfo->pRes->info.version)) {
- printDataBlock(pInfo->pRes, "stream scan ignore");
- blockDataCleanup(pInfo->pRes);
- continue;
- }
-
if (pInfo->pCreateTbRes->info.rows > 0) {
pInfo->scanMode = STREAM_SCAN_FROM_RES;
return pInfo->pCreateTbRes;
@@ -2515,6 +2482,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pInfo->igCheckUpdate = pTableScanNode->igCheckUpdate;
pInfo->igExpired = pTableScanNode->igExpired;
pInfo->twAggSup.maxTs = INT64_MIN;
+ pInfo->pState = NULL;
// todo(liuyao) get buff from rocks db;
void* buff = NULL;
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index af14b7096eebb0890c07551b40e6c0199bc41de9..2a0bfe4856c3ea04e487d18d1239f3c96314d9cc 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -13,7 +13,6 @@
* along with this program. If not, see .
*/
#include "executorimpl.h"
-#include "tglobal.h"
#include "filter.h"
#include "function.h"
#include "functionMgt.h"
@@ -21,12 +20,14 @@
#include "tcompare.h"
#include "tdatablock.h"
#include "tfill.h"
+#include "tglobal.h"
#include "tlog.h"
#include "ttime.h"
#define IS_FINAL_OP(op) ((op)->isFinal)
#define DEAULT_DELETE_MARK (1000LL * 60LL * 60LL * 24LL * 365LL * 10LL);
+
typedef struct SSessionAggOperatorInfo {
SOptrBasicInfo binfo;
SAggSupporter aggSup;
@@ -1367,12 +1368,12 @@ static void doDeleteWindows(SOperatorInfo* pOperator, SInterval* pInterval, SSDa
}
uint64_t winGpId = pGpDatas[i];
SWinKey winRes = {.ts = win.skey, .groupId = winGpId};
- void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinKey));
+ void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinKey));
if (chIds) {
getNextTimeWindow(pInterval, pInterval->precision, TSDB_ORDER_ASC, &win);
continue;
}
- bool res = doDeleteWindow(pOperator, win.skey, winGpId);
+ bool res = doDeleteWindow(pOperator, win.skey, winGpId);
if (pUpWins && res) {
taosArrayPush(pUpWins, &winRes);
}
@@ -1454,44 +1455,6 @@ STimeWindow getFinalTimeWindow(int64_t ts, SInterval* pInterval) {
return w;
}
-static void deleteIntervalDiscBuf(SStreamState* pState, SHashObj* pPullDataMap, TSKEY mark, SInterval* pInterval,
- SWinKey* key) {
- STimeWindow tw = getFinalTimeWindow(key->ts, pInterval);
- SWinKey next = {0};
- while (tw.ekey < mark) {
- SStreamStateCur* pCur = streamStateSeekKeyNext(pState, key);
- int32_t code = streamStateGetKVByCur(pCur, &next, NULL, 0);
- streamStateFreeCur(pCur);
-
- void* chIds = taosHashGet(pPullDataMap, key, sizeof(SWinKey));
- if (chIds && pPullDataMap) {
- SArray* chAy = *(SArray**)chIds;
- int32_t size = taosArrayGetSize(chAy);
- qDebug("===stream===window %" PRId64 " wait child size:%d", key->ts, size);
- for (int32_t i = 0; i < size; i++) {
- qDebug("===stream===window %" PRId64 " wait child id:%d", key->ts, *(int32_t*)taosArrayGet(chAy, i));
- }
- break;
- }
- qDebug("===stream===delete window %" PRId64, key->ts);
- int32_t codeDel = streamStateDel(pState, key);
- if (codeDel != TSDB_CODE_SUCCESS) {
- code = streamStateGetFirst(pState, key);
- if (code != TSDB_CODE_SUCCESS) {
- qDebug("===stream===stream state first key: empty-empty");
- return;
- }
- continue;
- }
- if (code == TSDB_CODE_SUCCESS) {
- *key = next;
- tw = getFinalTimeWindow(key->ts, pInterval);
- } else {
- break;
- }
- }
-}
-
static void closeChildIntervalWindow(SOperatorInfo* pOperator, SArray* pChildren, TSKEY maxTs) {
int32_t size = taosArrayGetSize(pChildren);
for (int32_t i = 0; i < size; i++) {
@@ -1579,14 +1542,6 @@ void destroyStreamFinalIntervalOperatorInfo(void* param) {
streamFileStateDestroy(pInfo->pState->pFileState);
taosMemoryFreeClear(pInfo->pState);
- if (pInfo->pChildren) {
- int32_t size = taosArrayGetSize(pInfo->pChildren);
- for (int32_t i = 0; i < size; i++) {
- SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
- destroyOperatorInfo(pChildOp);
- }
- taosArrayDestroy(pInfo->pChildren);
- }
nodesDestroyNode((SNode*)pInfo->pPhyNode);
colDataDestroy(&pInfo->twAggSup.timeWindowData);
pInfo->groupResInfo.pRows = taosArrayDestroy(pInfo->groupResInfo.pRows);
@@ -1658,20 +1613,20 @@ static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SInt
return needed;
}
-void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSupporter* pSup, SInterval* pInterval,
- STimeWindowAggSupp* pTwSup) {
+void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SStreamIntervalOperatorInfo* pInfo) {
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
- initIntervalDownStream(downstream->pDownstream[0], type, pSup, pInterval, pTwSup);
+ initIntervalDownStream(downstream->pDownstream[0], type, pInfo);
return;
}
SStreamScanInfo* pScanInfo = downstream->info;
pScanInfo->windowSup.parentType = type;
- pScanInfo->windowSup.pIntervalAggSup = pSup;
+ pScanInfo->windowSup.pIntervalAggSup = &pInfo->aggSup;
if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
- pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
+ pScanInfo->pUpdateInfo = updateInfoInitP(&pInfo->interval, pInfo->twAggSup.waterMark);
}
- pScanInfo->interval = *pInterval;
- pScanInfo->twAggSup = *pTwSup;
+ pScanInfo->interval = pInfo->interval;
+ pScanInfo->twAggSup = pInfo->twAggSup;
+ pScanInfo->pState = pInfo->pState;
}
void initStreamFunciton(SqlFunctionCtx* pCtx, int32_t numOfExpr) {
@@ -2096,12 +2051,11 @@ void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int3
}
}
-bool hasIntervalWindow(SStreamState* pState, SWinKey* pKey) {
- return streamStateCheck(pState, pKey);
-}
+bool hasIntervalWindow(SStreamState* pState, SWinKey* pKey) { return streamStateCheck(pState, pKey); }
int32_t setIntervalOutputBuf(SStreamState* pState, STimeWindow* win, SRowBuffPos** pResult, int64_t groupId,
- SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
+ SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset,
+ SAggSupporter* pAggSup) {
SWinKey key = {
.ts = win->skey,
.groupId = groupId,
@@ -2115,68 +2069,15 @@ int32_t setIntervalOutputBuf(SStreamState* pState, STimeWindow* win, SRowBuffPos
*pResult = (SRowBuffPos*)value;
SResultRow* res = (SResultRow*)((*pResult)->pRowBuff);
// set time window for current result
- res-> win = (*win);
+ res->win = (*win);
setResultRowInitCtx(res, pCtx, numOfOutput, rowEntryInfoOffset);
return TSDB_CODE_SUCCESS;
}
-static void rebuildIntervalWindow(SOperatorInfo* pOperator, SArray* pWinArray, SSHashObj* pUpdatedMap) {
- SStreamIntervalOperatorInfo* pInfo = pOperator->info;
- SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
- int32_t size = taosArrayGetSize(pWinArray);
- int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
- SExprSupp* pSup = &pOperator->exprSupp;
- if (!pInfo->pChildren) {
- return;
- }
- for (int32_t i = 0; i < size; i++) {
- SWinKey* pWinRes = taosArrayGet(pWinArray, i);
- SRowBuffPos* pCurResPos = NULL;
- SResultRow* pCurResult = NULL;
- STimeWindow parentWin = getFinalTimeWindow(pWinRes->ts, &pInfo->interval);
- if (isDeletedStreamWindow(&parentWin, pWinRes->groupId, pInfo->pState, &pInfo->twAggSup)) {
- continue;
- }
-
- int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
- int32_t num = 0;
- for (int32_t j = 0; j < numOfChildren; j++) {
- SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, j);
- SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
- SExprSupp* pChildSup = &pChildOp->exprSupp;
- if (!hasIntervalWindow(pChInfo->pState, pWinRes)) {
- continue;
- }
- if (num == 0) {
- int32_t code = setIntervalOutputBuf(pInfo->pState, &parentWin, &pCurResPos, pWinRes->groupId, pSup->pCtx, numOfOutput,
- pSup->rowEntryInfoOffset, &pInfo->aggSup);
- ASSERT(pCurResPos != NULL);
- pCurResult = (SResultRow*) pCurResPos->pRowBuff;
- if (code != TSDB_CODE_SUCCESS || pCurResult == NULL) {
- T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
- }
-
- }
- num++;
- SRowBuffPos* pChResPos = NULL;
- SResultRow* pChResult = NULL;
- setIntervalOutputBuf(pChInfo->pState, &parentWin, &pChResPos, pWinRes->groupId, pChildSup->pCtx, pChildSup->numOfExprs,
- pChildSup->rowEntryInfoOffset, &pChInfo->aggSup);
- pChResult = (SResultRow*) pChResPos->pRowBuff;
- updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &parentWin, true);
- compactFunctions(pSup->pCtx, pChildSup->pCtx, numOfOutput, pTaskInfo, &pInfo->twAggSup.timeWindowData);
- }
- if (num > 0 && pUpdatedMap) {
- saveWinResultInfo(pCurResult->win.skey, pWinRes->groupId, pCurResPos, pUpdatedMap);
- saveOutputBuf(pInfo->pState, pWinRes, pCurResult, pInfo->aggSup.resultRowSize);
- }
- }
-}
-
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup) {
if (pTwSup->maxTs != INT64_MIN && pWin->ekey < pTwSup->maxTs - pTwSup->deleteMark) {
SWinKey key = {.ts = pWin->skey, .groupId = groupId};
- if (streamStateCheck(pState, &key)) {
+ if (!hasIntervalWindow(pState, &key)) {
return true;
}
return false;
@@ -2290,9 +2191,8 @@ static void addRetriveWindow(SArray* wins, SStreamIntervalOperatorInfo* pInfo) {
.window = nextWin, .groupId = winKey->groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey};
// add pull data request
if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
- int32_t size1 = taosArrayGetSize(pInfo->pChildren);
- addPullWindow(pInfo->pPullDataMap, winKey, size1);
- qDebug("===stream===prepare retrive for delete %" PRId64 ", size:%d", winKey->ts, size1);
+ addPullWindow(pInfo->pPullDataMap, winKey, pInfo->numOfChild);
+ qDebug("===stream===prepare retrive for delete %" PRId64 ", size:%d", winKey->ts, pInfo->numOfChild);
}
}
}
@@ -2412,7 +2312,7 @@ static int32_t getNextQualifiedFinalWindow(SInterval* pInterval, STimeWindow* pN
return startPos;
}
-static void setStreamDataVersion(SExecTaskInfo* pTaskInfo, int64_t version, int64_t ckId) {
+static void setStreamDataVersion(SExecTaskInfo* pTaskInfo, int64_t version, int64_t ckId) {
pTaskInfo->streamInfo.dataVersion = version;
pTaskInfo->streamInfo.checkPointId = ckId;
}
@@ -2453,7 +2353,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
continue;
}
- if (IS_FINAL_OP(pInfo) && pInfo->pChildren) {
+ if (IS_FINAL_OP(pInfo) && pInfo->numOfChild > 0) {
bool ignore = true;
SWinKey winRes = {
.ts = nextWin.skey,
@@ -2465,9 +2365,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
.window = nextWin, .groupId = groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey};
// add pull data request
if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
- int32_t size = taosArrayGetSize(pInfo->pChildren);
- addPullWindow(pInfo->pPullDataMap, &winRes, size);
- qDebug("===stream===prepare retrive %" PRId64 ", size:%d", winRes.ts, size);
+ addPullWindow(pInfo->pPullDataMap, &winRes, pInfo->numOfChild);
}
} else {
int32_t index = -1;
@@ -2484,7 +2382,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
}
if (ignore) {
- startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
+ startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, startPos);
if (startPos < 0) {
break;
}
@@ -2493,8 +2391,8 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
}
int32_t code = setIntervalOutputBuf(pInfo->pState, &nextWin, &pResPos, groupId, pSup->pCtx, numOfOutput,
- pSup->rowEntryInfoOffset, &pInfo->aggSup);
- pResult = (SResultRow*) pResPos->pRowBuff;
+ pSup->rowEntryInfoOffset, &pInfo->aggSup);
+ pResult = (SResultRow*)pResPos->pRowBuff;
if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
}
@@ -2506,8 +2404,8 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
}
SWinKey key = {
- .ts = pResult->win.skey,
- .groupId = groupId,
+ .ts = pResult->win.skey,
+ .groupId = groupId,
};
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) {
saveWinResult(&key, pResPos, pUpdatedMap);
@@ -2554,8 +2452,8 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p
static inline int winPosCmprImpl(const void* pKey1, const void* pKey2) {
SRowBuffPos* pos1 = *(SRowBuffPos**)pKey1;
SRowBuffPos* pos2 = *(SRowBuffPos**)pKey2;
- SWinKey* pWin1 = (SWinKey*)pos1->pKey;
- SWinKey* pWin2 = (SWinKey*)pos2->pKey;
+ SWinKey* pWin1 = (SWinKey*)pos1->pKey;
+ SWinKey* pWin2 = (SWinKey*)pos2->pKey;
if (pWin1->groupId > pWin2->groupId) {
return 1;
@@ -2611,10 +2509,9 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
setStreamDataVersion(pTaskInfo, pInfo->dataVersion, pInfo->pState->checkPointId);
qDebug("===stream===clear semi operator");
} else {
- deleteIntervalDiscBuf(pInfo->pState, pInfo->pPullDataMap, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark,
- &pInfo->interval, &pInfo->delKey);
if (pInfo->twAggSup.maxTs - pInfo->twAggSup.checkPointInterval > pInfo->twAggSup.checkPointTs) {
streamStateCommit(pInfo->pState);
+ streamStateDeleteCheckPoint(pInfo->pState, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark);
pInfo->twAggSup.checkPointTs = pInfo->twAggSup.maxTs;
}
qDebug("===stream===interval final close");
@@ -2757,7 +2654,7 @@ int64_t getDeleteMark(SIntervalPhysiNode* pIntervalPhyNode) {
}
TSKEY compareTs(void* pKey) {
- SWinKey* pWinKey = (SWinKey*) pKey;
+ SWinKey* pWinKey = (SWinKey*)pKey;
return pWinKey->ts;
}
@@ -2786,7 +2683,8 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
.deleteMarkSaved = 0,
.calTriggerSaved = 0,
.checkPointTs = 0,
- .checkPointInterval = convertTimePrecision(tsCheckpointInterval, TSDB_TIME_PRECISION_MILLI, pInfo->interval.precision),
+ .checkPointInterval =
+ convertTimePrecision(tsCheckpointInterval, TSDB_TIME_PRECISION_MILLI, pInfo->interval.precision),
};
ASSERTS(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay");
pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
@@ -2821,24 +2719,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
streamStateSetNumber(pInfo->pState, -1);
initResultRowInfo(&pInfo->binfo.resultRowInfo);
- pInfo->pChildren = NULL;
- if (numOfChild > 0) {
- pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
- if (!pInfo->pChildren) {
- goto _error;
- }
- for (int32_t i = 0; i < numOfChild; i++) {
- SOperatorInfo* pChildOp = createStreamFinalIntervalOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
- if (pChildOp) {
- SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
- pChInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
- taosArrayPush(pInfo->pChildren, &pChildOp);
- streamStateSetNumber(pChInfo->pState, i);
- continue;
- }
- goto _error;
- }
- }
+ pInfo->numOfChild = numOfChild;
pInfo->pPhyNode = (SPhysiNode*)nodesCloneNode((SNode*)pPhyNode);
@@ -2869,8 +2750,8 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pInfo->numOfDatapack = 0;
pInfo->pUpdated = NULL;
pInfo->pUpdatedMap = NULL;
- pInfo->pState->pFileState = streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, compareTs,
- pInfo->pState, pInfo->twAggSup.deleteMark);
+ pInfo->pState->pFileState = streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize,
+ compareTs, pInfo->pState, pInfo->twAggSup.deleteMark);
pInfo->dataVersion = 0;
pOperator->operatorType = pPhyNode->type;
@@ -2881,7 +2762,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pOperator->fpSet = createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, destroyStreamFinalIntervalOperatorInfo,
optrDefaultBufFn, NULL);
if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL) {
- initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
+ initIntervalDownStream(downstream, pPhyNode->type, pInfo);
}
code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) {
@@ -3378,13 +3259,13 @@ static void rebuildSessionWindow(SOperatorInfo* pOperator, SArray* pWinArray, SS
SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
int32_t numOfOutput = pSup->numOfExprs;
- int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
+ int32_t numOfChild = taosArrayGetSize(pInfo->pChildren);
for (int32_t i = 0; i < size; i++) {
SSessionKey* pWinKey = taosArrayGet(pWinArray, i);
int32_t num = 0;
SResultWindowInfo parentWin = {0};
- for (int32_t j = 0; j < numOfChildren; j++) {
+ for (int32_t j = 0; j < numOfChild; j++) {
SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, j);
SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
SStreamAggSupporter* pChAggSup = &pChInfo->streamAggSup;
@@ -4849,11 +4730,10 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes;
}
- deleteIntervalDiscBuf(pInfo->pState, NULL, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark, &pInfo->interval,
- &pInfo->delKey);
setOperatorCompleted(pOperator);
if (pInfo->twAggSup.maxTs - pInfo->twAggSup.checkPointInterval > pInfo->twAggSup.checkPointTs) {
streamStateCommit(pInfo->pState);
+ streamStateDeleteCheckPoint(pInfo->pState, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark);
setStreamDataVersion(pTaskInfo, pInfo->dataVersion, pInfo->pState->checkPointId);
pInfo->twAggSup.checkPointTs = pInfo->twAggSup.maxTs;
}
@@ -4971,23 +4851,24 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
- pInfo->interval = (SInterval) {
- .interval = pIntervalPhyNode->interval,
- .sliding = pIntervalPhyNode->sliding,
- .intervalUnit = pIntervalPhyNode->intervalUnit,
- .slidingUnit = pIntervalPhyNode->slidingUnit,
- .offset = pIntervalPhyNode->offset,
- .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision,
+ pInfo->interval = (SInterval){
+ .interval = pIntervalPhyNode->interval,
+ .sliding = pIntervalPhyNode->sliding,
+ .intervalUnit = pIntervalPhyNode->intervalUnit,
+ .slidingUnit = pIntervalPhyNode->slidingUnit,
+ .offset = pIntervalPhyNode->offset,
+ .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision,
};
- pInfo->twAggSup = (STimeWindowAggSupp) {
+ pInfo->twAggSup = (STimeWindowAggSupp){
.waterMark = pIntervalPhyNode->window.watermark,
.calTrigger = pIntervalPhyNode->window.triggerType,
.maxTs = INT64_MIN,
.minTs = INT64_MAX,
.deleteMark = getDeleteMark(pIntervalPhyNode),
.checkPointTs = 0,
- .checkPointInterval = convertTimePrecision(tsCheckpointInterval, TSDB_TIME_PRECISION_MILLI, pInfo->interval.precision),
+ .checkPointInterval =
+ convertTimePrecision(tsCheckpointInterval, TSDB_TIME_PRECISION_MILLI, pInfo->interval.precision),
};
ASSERTS(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay");
@@ -5038,21 +4919,21 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->pullIndex = 0;
pInfo->pPullDataRes = NULL;
pInfo->isFinal = false;
- pInfo->pChildren = NULL;
+ pInfo->numOfChild = 0;
pInfo->delKey.ts = INT64_MAX;
pInfo->delKey.groupId = 0;
pInfo->numOfDatapack = 0;
pInfo->pUpdated = NULL;
pInfo->pUpdatedMap = NULL;
- pInfo->pState->pFileState = streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, compareTs,
- pInfo->pState, pInfo->twAggSup.deleteMark);
+ pInfo->pState->pFileState = streamFileStateInit(tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize,
+ compareTs, pInfo->pState, pInfo->twAggSup.deleteMark);
setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamIntervalAgg, NULL,
destroyStreamFinalIntervalOperatorInfo, optrDefaultBufFn, NULL);
- initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
+ initIntervalDownStream(downstream, pPhyNode->type, pInfo);
code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
diff --git a/source/libs/stream/inc/streamBackendRocksdb.h b/source/libs/stream/inc/streamBackendRocksdb.h
index a06aef365f19579b478972b07b0786e213d33a59..a33a0a577b7e06fc39bc979498ce7a6b6e943360 100644
--- a/source/libs/stream/inc/streamBackendRocksdb.h
+++ b/source/libs/stream/inc/streamBackendRocksdb.h
@@ -89,4 +89,5 @@ int32_t streamStatePutBatch_rocksdb(SStreamState* pState, void* pBatch);
int32_t streamDefaultPut_rocksdb(SStreamState* pState, const void* key, void* pVal, int32_t pVLen);
int32_t streamDefaultGet_rocksdb(SStreamState* pState, const void* key, void** pVal, int32_t* pVLen);
int32_t streamDefaultDel_rocksdb(SStreamState* pState, const void* key);
+int32_t streamDefaultIter_rocksdb(SStreamState* pState, const void* start, const void* end, SArray* result);
#endif
\ No newline at end of file
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index a69a86e400ba087e0d21706e09de5f470754ed1f..fee65d6c58e95097f3d6e7614252da97f67934a5 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -1066,12 +1066,21 @@ void streamStateDestroy(SStreamState* pState) {
#ifdef USE_ROCKSDB
streamFileStateDestroy(pState->pFileState);
streamStateDestroy_rocksdb(pState);
+ taosMemoryFreeClear(pState->parNameMap);
// do nothong
#endif
taosMemoryFreeClear(pState->pTdbState);
taosMemoryFreeClear(pState);
}
+int32_t streamStateDeleteCheckPoint(SStreamState* pState, TSKEY mark) {
+#ifdef USE_ROCKSDB
+ return deleteExpiredCheckPoint(pState->pFileState, mark);
+#else
+ return 0;
+#endif
+}
+
#if 0
char* streamStateSessionDump(SStreamState* pState) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
diff --git a/source/libs/stream/src/streamStateRocksdb.c b/source/libs/stream/src/streamStateRocksdb.c
index e700e8b97a5f1e88d366613f05b65863afdf7cb6..9977732a9aae0fe3d16c38ba6fdb2ee164888279 100644
--- a/source/libs/stream/src/streamStateRocksdb.c
+++ b/source/libs/stream/src/streamStateRocksdb.c
@@ -13,13 +13,22 @@
* along with this program. If not, see .
*/
+#include "rocksdb/c.h"
#include "streamBackendRocksdb.h"
#include "tcommon.h"
-#include "tlog.h"
int defaultKeyComp(void* state, const char* aBuf, size_t aLen, const char* bBuf, size_t bLen) {
- //
- return memcmp(aBuf, bBuf, aLen);
+ int ret = memcmp(aBuf, bBuf, aLen);
+ if (ret == 0) {
+ if (aLen < bLen)
+ return -1;
+ else if (aLen > bLen)
+ return 1;
+ else
+ return 0;
+ } else {
+ return ret;
+ }
}
int defaultKeyEncode(void* k, char* buf) {
int len = strlen((char*)k);
@@ -280,6 +289,10 @@ int parKeyToString(void* k, char* buf) {
return n;
}
+typedef struct {
+ void* tableOpt;
+ void* lru; // global or not
+} rocksdbCfParam;
const char* cfName[] = {"default", "state", "fill", "sess", "func", "parname", "partag"};
typedef int (*EncodeFunc)(void* key, char* buf);
@@ -297,6 +310,8 @@ const char* compareFuncKeyName(void* name);
const char* compareParKeyName(void* name);
const char* comparePartagKeyName(void* name);
+void destroyFunc(void* stata) { return; }
+
typedef struct {
const char* key;
int32_t len;
@@ -312,14 +327,19 @@ typedef struct {
SCfInit ginitDict[] = {
{"default", strlen("default"), 0, defaultKeyComp, defaultKeyEncode, defaultKeyDecode, defaultKeyToString,
- compareDefaultName},
- {"state", strlen("state"), 1, stateKeyDBComp, stateKeyEncode, stateKeyDecode, stateKeyToString, compareStateName},
- {"fill", strlen("fill"), 2, winKeyDBComp, winKeyEncode, winKeyDecode, winKeyToString, compareWinKeyName},
+ compareDefaultName, destroyFunc},
+ {"state", strlen("state"), 1, stateKeyDBComp, stateKeyEncode, stateKeyDecode, stateKeyToString, compareStateName,
+ destroyFunc},
+ {"fill", strlen("fill"), 2, winKeyDBComp, winKeyEncode, winKeyDecode, winKeyToString, compareWinKeyName,
+ destroyFunc},
{"sess", strlen("sess"), 3, stateSessionKeyDBComp, stateSessionKeyEncode, stateSessionKeyDecode,
- stateSessionKeyToString, compareSessionKeyName},
- {"func", strlen("func"), 4, tupleKeyDBComp, tupleKeyEncode, tupleKeyDecode, tupleKeyToString, compareFuncKeyName},
- {"parname", strlen("parname"), 5, parKeyDBComp, parKeyEncode, parKeyDecode, parKeyToString, compareParKeyName},
- {"partag", strlen("partag"), 6, parKeyDBComp, parKeyEncode, parKeyDecode, parKeyToString, comparePartagKeyName},
+ stateSessionKeyToString, compareSessionKeyName, destroyFunc},
+ {"func", strlen("func"), 4, tupleKeyDBComp, tupleKeyEncode, tupleKeyDecode, tupleKeyToString, compareFuncKeyName,
+ destroyFunc},
+ {"parname", strlen("parname"), 5, parKeyDBComp, parKeyEncode, parKeyDecode, parKeyToString, compareParKeyName,
+ destroyFunc},
+ {"partag", strlen("partag"), 6, parKeyDBComp, parKeyEncode, parKeyDecode, parKeyToString, comparePartagKeyName,
+ destroyFunc},
};
const char* compareDefaultName(void* name) { return ginitDict[0].key; }
@@ -330,8 +350,6 @@ const char* compareFuncKeyName(void* name) { return ginitDict[4].key; }
const char* compareParKeyName(void* name) { return ginitDict[5].key; }
const char* comparePartagKeyName(void* name) { return ginitDict[6].key; }
-void destroyFunc(void* stata) { return; }
-
int streamInitBackend(SStreamState* pState, char* path) {
rocksdb_env_t* env = rocksdb_create_default_env(); // rocksdb_envoptions_create();
rocksdb_env_set_low_priority_background_threads(env, 15);
@@ -349,6 +367,7 @@ int streamInitBackend(SStreamState* pState, char* path) {
char* err = NULL;
int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
+ rocksdbCfParam* param = taosMemoryCalloc(cfLen, sizeof(rocksdbCfParam));
const rocksdb_options_t** cfOpt = taosMemoryCalloc(cfLen, sizeof(rocksdb_options_t*));
for (int i = 0; i < cfLen; i++) {
cfOpt[i] = rocksdb_options_create_copy(opts);
@@ -362,6 +381,8 @@ int streamInitBackend(SStreamState* pState, char* path) {
rocksdb_options_set_block_based_table_factory((rocksdb_options_t*)cfOpt[i], tableOpt);
+ param[i].tableOpt = tableOpt;
+ param[i].lru = cache;
// rocksdb_slicetransform_t* trans = rocksdb_slicetransform_create_fixed_prefix(8);
// rocksdb_options_set_prefix_extractor((rocksdb_options_t*)cfOpt[i], trans);
};
@@ -386,6 +407,8 @@ int streamInitBackend(SStreamState* pState, char* path) {
pState->pTdbState->cfOpts = (rocksdb_options_t**)cfOpt;
pState->pTdbState->pCompare = pCompare;
pState->pTdbState->dbOpt = opts;
+ pState->pTdbState->param = param;
+ pState->pTdbState->env = env;
return 0;
}
void streamCleanBackend(SStreamState* pState) {
@@ -393,12 +416,17 @@ void streamCleanBackend(SStreamState* pState) {
qInfo("rocksdb already free");
return;
}
- int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
+ int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
+ rocksdbCfParam* param = pState->pTdbState->param;
for (int i = 0; i < cfLen; i++) {
rocksdb_column_family_handle_destroy(pState->pTdbState->pHandle[i]);
rocksdb_options_destroy(pState->pTdbState->cfOpts[i]);
rocksdb_comparator_destroy(pState->pTdbState->pCompare[i]);
+
+ rocksdb_cache_destroy(param[i].lru);
+ rocksdb_block_based_options_destroy(param[i].tableOpt);
}
+ taosMemoryFree(pState->pTdbState->param);
rocksdb_options_destroy(pState->pTdbState->dbOpt);
taosMemoryFreeClear(pState->pTdbState->pHandle);
@@ -412,6 +440,7 @@ void streamCleanBackend(SStreamState* pState) {
pState->pTdbState->readOpts = NULL;
rocksdb_close(pState->pTdbState->rocksdb);
+ rocksdb_env_destroy(pState->pTdbState->env);
pState->pTdbState->rocksdb = NULL;
}
@@ -438,9 +467,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
rocksdb_readoptions_t** readOpt) {
int idx = streamGetInit(cfName);
- //*snapshot = (rocksdb_snapshot_t*)rocksdb_create_snapshot(pState->pTdbState->rocksdb);
if (snapshot != NULL) {
- *snapshot = NULL;
+ *snapshot = (rocksdb_snapshot_t*)rocksdb_create_snapshot(pState->pTdbState->rocksdb);
}
rocksdb_readoptions_t* rOpt = rocksdb_readoptions_create();
@@ -618,6 +646,39 @@ int32_t streamDefaultDel_rocksdb(SStreamState* pState, const void* key) {
return code;
}
+int32_t streamDefaultIter_rocksdb(SStreamState* pState, const void* start, const void* end, SArray* result) {
+ int code = 0;
+ char* err = NULL;
+
+ rocksdb_snapshot_t* snapshot = NULL;
+ rocksdb_readoptions_t* readopts = NULL;
+ rocksdb_iterator_t* pIter = streamStateIterCreate(pState, "default", &snapshot, &readopts);
+ if (pIter == NULL) {
+ return -1;
+ }
+
+ rocksdb_iter_seek(pIter, start, strlen(start));
+ while (rocksdb_iter_valid(pIter)) {
+ const char* key = rocksdb_iter_key(pIter, NULL);
+ if (end != NULL && strcmp(key, end) > 0) {
+ break;
+ }
+ if (strncmp(key, start, strlen(start)) == 0 && strlen(key) >= strlen(start) + 1) {
+ int64_t checkPoint = 0;
+ if (sscanf(key + strlen(key), ":%" PRId64 "", &checkPoint) == 1) {
+ taosArrayPush(result, &checkPoint);
+ }
+ } else {
+ break;
+ }
+ rocksdb_iter_next(pIter);
+ }
+ rocksdb_release_snapshot(pState->pTdbState->rocksdb, snapshot);
+ rocksdb_readoptions_destroy(readopts);
+ rocksdb_iter_destroy(pIter);
+ return code;
+}
+
int32_t streamStateGet_rocksdb(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
int code = 0;
SStateKey sKey = {.key = *key, .opNum = pState->number};
diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c
index be12c72d004711e6c126792b5a7cfe4053ad1f18..70a1c543f648104064c311084b5d634517be4c2e 100644
--- a/source/libs/stream/src/streamUpdate.c
+++ b/source/libs/stream/src/streamUpdate.c
@@ -128,9 +128,7 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma
pInfo->pCloseWinSBF = NULL;
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);
pInfo->pMap = taosHashInit(DEFAULT_MAP_CAPACITY, hashFn, true, HASH_NO_LOCK);
- pInfo->maxVersion = 0;
- pInfo->scanGroupId = 0;
- pInfo->scanWindow = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
+ pInfo->maxDataVersion = 0;
return pInfo;
}
@@ -242,29 +240,6 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts) {
return true;
}
-void updateInfoSetScanRange(SUpdateInfo *pInfo, STimeWindow *pWin, uint64_t groupId, uint64_t version) {
- qDebug("===stream===groupId:%" PRIu64 ", startTs:%" PRIu64 ", endTs:%" PRIu64 ", version:%" PRIu64, groupId,
- pWin->skey, pWin->ekey, version);
- pInfo->scanWindow = *pWin;
- pInfo->scanGroupId = groupId;
- pInfo->maxVersion = version;
-}
-
-bool updateInfoIgnore(SUpdateInfo *pInfo, STimeWindow *pWin, uint64_t groupId, uint64_t version) {
- if (!pInfo) {
- return false;
- }
- qDebug("===stream===check groupId:%" PRIu64 ", startTs:%" PRIu64 ", endTs:%" PRIu64 ", version:%" PRIu64, groupId,
- pWin->skey, pWin->ekey, version);
- if (pInfo->scanGroupId == groupId && pInfo->scanWindow.skey <= pWin->skey && pWin->ekey <= pInfo->scanWindow.ekey &&
- version <= pInfo->maxVersion) {
- qDebug("===stream===ignore groupId:%" PRIu64 ", startTs:%" PRIu64 ", endTs:%" PRIu64 ", version:%" PRIu64, groupId,
- pWin->skey, pWin->ekey, version);
- return true;
- }
- return false;
-}
-
void updateInfoDestroy(SUpdateInfo *pInfo) {
if (pInfo == NULL) {
return;
@@ -337,10 +312,7 @@ int32_t updateInfoSerialize(void *buf, int32_t bufLen, const SUpdateInfo *pInfo)
if (tEncodeI64(&encoder, *(TSKEY *)pIte) < 0) return -1;
}
- if (tEncodeI64(&encoder, pInfo->scanWindow.skey) < 0) return -1;
- if (tEncodeI64(&encoder, pInfo->scanWindow.ekey) < 0) return -1;
- if (tEncodeU64(&encoder, pInfo->scanGroupId) < 0) return -1;
- if (tEncodeU64(&encoder, pInfo->maxVersion) < 0) return -1;
+ if (tEncodeU64(&encoder, pInfo->maxDataVersion) < 0) return -1;
tEndEncode(&encoder);
@@ -393,11 +365,7 @@ int32_t updateInfoDeserialize(void *buf, int32_t bufLen, SUpdateInfo *pInfo) {
taosHashPut(pInfo->pMap, &uid, sizeof(uint64_t), &ts, sizeof(TSKEY));
}
ASSERT(mapSize == taosHashGetSize(pInfo->pMap));
-
- if (tDecodeI64(&decoder, &pInfo->scanWindow.skey) < 0) return -1;
- if (tDecodeI64(&decoder, &pInfo->scanWindow.ekey) < 0) return -1;
- if (tDecodeU64(&decoder, &pInfo->scanGroupId) < 0) return -1;
- if (tDecodeU64(&decoder, &pInfo->maxVersion) < 0) return -1;
+ if (tDecodeU64(&decoder, &pInfo->maxDataVersion) < 0) return -1;
tEndDecode(&decoder);
diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c
index dedf2b114f17ca2fe37d4ec692c32194f323afe1..88fdfb6b3c6694608847c19291e6ad1bce3e59c1 100644
--- a/source/libs/stream/src/tstreamFileState.c
+++ b/source/libs/stream/src/tstreamFileState.c
@@ -313,15 +313,13 @@ SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {
return pFileState->usedBuffs;
}
-void streamFileStateDecode(SStreamFileState* pFileState, void* pBuff, int32_t len) {
- pBuff = taosDecodeFixedI64(pBuff, &pFileState->flushMark);
-}
+void streamFileStateDecode(TSKEY* key, void* pBuff, int32_t len) { pBuff = taosDecodeFixedI64(pBuff, key); }
-void streamFileStateEncode(SStreamFileState* pFileState, void** pVal, int32_t* pLen) {
+void streamFileStateEncode(TSKEY* key, void** pVal, int32_t* pLen) {
*pLen = sizeof(TSKEY);
(*pVal) = taosMemoryCalloc(1, *pLen);
void* buff = *pVal;
- taosEncodeFixedI64(&buff, pFileState->flushMark);
+ taosEncodeFixedI64(&buff, *key);
}
int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {
@@ -347,31 +345,88 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
if (streamStateGetBatchSize(batch) > 0) {
code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
}
+ streamStateClearBatch(batch);
if (flushState) {
- int32_t len = 0;
- void* buff = NULL;
- streamFileStateEncode(pFileState, &buff, &len);
- SWinKey key = {.ts = -1, .groupId = 0}; // dengyihao
- streamStatePut_rocksdb(pFileState->pFileStore, &key, buff, len);
- taosMemoryFree(buff);
+ const char* taskKey = "streamFileState";
+ {
+ char keyBuf[128] = {0};
+ void* valBuf = NULL;
+ int32_t len = 0;
+ sprintf(keyBuf, "%s:%" PRId64 "", taskKey, ((SStreamState*)pFileState->pFileStore)->checkPointId);
+ streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);
+ code = streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len);
+ taosMemoryFree(valBuf);
+ }
+ {
+ char keyBuf[128] = {0};
+ char valBuf[64] = {0};
+ int32_t len = 0;
+ memcpy(keyBuf, taskKey, strlen(taskKey));
+ len = sprintf(valBuf, "%" PRId64 "", ((SStreamState*)pFileState->pFileStore)->checkPointId);
+ code = streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len);
+ }
+ streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
}
-
streamStateDestroyBatch(batch);
+
+ return code;
+}
+
+int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {
+ const char* taskKey = "streamFileState";
+ char keyBuf[128] = {0};
+ sprintf(keyBuf, "%s:%" PRId64 "", taskKey, checkpointId);
+ return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);
+}
+
+int32_t getSnapshotIdList(SStreamFileState* pFileState, SArray* list) {
+ const char* taskKey = "streamFileState";
+ return streamDefaultIter_rocksdb(pFileState->pFileStore, taskKey, NULL, list);
+}
+
+int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
+ int32_t code = TSDB_CODE_SUCCESS;
+ const char* taskKey = "streamFileState";
+ int64_t maxCheckPointId = 0;
+ {
+ char buf[128] = {0};
+ void* val = NULL;
+ int32_t len = 0;
+ memcpy(buf, taskKey, strlen(taskKey));
+ code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
+ if (code != 0) {
+ return TSDB_CODE_FAILED;
+ }
+ sscanf(val, "%" PRId64 "", &maxCheckPointId);
+ taosMemoryFree(val);
+ }
+ for (int64_t i = maxCheckPointId; i > 0; i--) {
+ char buf[128] = {0};
+ void* val = 0;
+ int32_t len = 0;
+ sprintf(buf, "%s:%" PRId64 "", taskKey, i);
+ code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
+ if (code != 0) {
+ return TSDB_CODE_FAILED;
+ }
+ TSKEY ts;
+ sscanf(val, "%" PRId64 "", &ts);
+ taosMemoryFree(val);
+ if (ts < mark) {
+ forceRemoveCheckpoint(pFileState, i);
+ break;
+ } else {
+ }
+ }
return code;
}
int32_t recoverSnapshot(SStreamFileState* pFileState) {
int32_t code = TSDB_CODE_SUCCESS;
- SWinKey stkey = {.ts = -1, .groupId = 0}; // dengyihao
+ deleteExpiredCheckPoint(pFileState, pFileState->maxTs - pFileState->deleteMark);
void* pStVal = NULL;
int32_t len = 0;
- code = streamStateGet_rocksdb(pFileState->pFileStore, &stkey, &pStVal, &len);
- if (code == TSDB_CODE_SUCCESS) {
- streamFileStateDecode(pFileState, pStVal, len);
- } else {
- return TSDB_CODE_FAILED;
- }
SWinKey key = {.groupId = 0, .ts = 0};
SStreamStateCur* pCur = streamStateGetCur_rocksdb(pFileState->pFileStore, &key);
diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index 2b5487c260e219de9c28823adb3fabf6ab0d0b74..d8ff2d4996cfcfdb5ab0e83e7c44eafc21133530 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -223,40 +223,47 @@
,,n,script,./test.sh -f tsim/stream/basic0.sim -g
,,y,script,./test.sh -f tsim/stream/basic1.sim
,,y,script,./test.sh -f tsim/stream/basic2.sim
+,,y,script,./test.sh -f tsim/stream/basic3.sim
+,,y,script,./test.sh -f tsim/stream/basic4.sim
+,,y,script,./test.sh -f tsim/stream/checkStreamSTable1.sim
+,,y,script,./test.sh -f tsim/stream/checkStreamSTable.sim
+,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
+,,y,script,./test.sh -f tsim/stream/deleteSession.sim
+,,y,script,./test.sh -f tsim/stream/deleteState.sim
+,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
+,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
+,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/drop_stream.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim
-,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
-,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
-,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
-,,y,script,./test.sh -f tsim/stream/session0.sim
-,,y,script,./test.sh -f tsim/stream/session1.sim
-,,y,script,./test.sh -f tsim/stream/state0.sim
-,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
-,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
-,,y,script,./test.sh -f tsim/stream/partitionby.sim
-,,y,script,./test.sh -f tsim/stream/partitionby1.sim
-,,y,script,./test.sh -f tsim/stream/schedSnode.sim
-,,y,script,./test.sh -f tsim/stream/windowClose.sim
-,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
-,,y,script,./test.sh -f tsim/stream/sliding.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
-,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
-,,y,script,./test.sh -f tsim/stream/deleteSession.sim
-,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalLinear.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
+,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext.sim
+,,y,script,./test.sh -f tsim/stream/fillIntervalRange.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalValue.sim
+,,y,script,./test.sh -f tsim/stream/ignoreCheckUpdate.sim
+,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
+,,y,script,./test.sh -f tsim/stream/partitionby1.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
+,,y,script,./test.sh -f tsim/stream/partitionby.sim
+,,y,script,./test.sh -f tsim/stream/schedSnode.sim
+,,y,script,./test.sh -f tsim/stream/session0.sim
+,,y,script,./test.sh -f tsim/stream/session1.sim
+,,y,script,./test.sh -f tsim/stream/sliding.sim
+,,y,script,./test.sh -f tsim/stream/state0.sim
+,,y,script,./test.sh -f tsim/stream/state1.sim
+,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
+,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag1.sim
-,,y,script,./test.sh -f tsim/stream/checkStreamSTable.sim
-,,y,script,./test.sh -f tsim/stream/checkStreamSTable1.sim
+,,y,script,./test.sh -f tsim/stream/udTableAndTag2.sim
+,,y,script,./test.sh -f tsim/stream/windowClose.sim
,,y,script,./test.sh -f tsim/trans/lossdata1.sim
,,y,script,./test.sh -f tsim/trans/create_db.sim
,,y,script,./test.sh -f tsim/tmq/basic1.sim
@@ -1336,38 +1343,47 @@
,,n,script,./test.sh -f tsim/stream/basic0.sim -g
,,y,script,./test.sh -f tsim/stream/basic1.sim
,,y,script,./test.sh -f tsim/stream/basic2.sim
+,,y,script,./test.sh -f tsim/stream/basic3.sim
+,,y,script,./test.sh -f tsim/stream/basic4.sim
+,,y,script,./test.sh -f tsim/stream/checkStreamSTable1.sim
+,,y,script,./test.sh -f tsim/stream/checkStreamSTable.sim
+,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
+,,y,script,./test.sh -f tsim/stream/deleteSession.sim
+,,y,script,./test.sh -f tsim/stream/deleteState.sim
+,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
+,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
+,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/drop_stream.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim
-,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
-,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
-,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
-,,y,script,./test.sh -f tsim/stream/session0.sim
-,,y,script,./test.sh -f tsim/stream/session1.sim
-,,y,script,./test.sh -f tsim/stream/state0.sim
-,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
-,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
-,,y,script,./test.sh -f tsim/stream/partitionby.sim
-,,y,script,./test.sh -f tsim/stream/partitionby1.sim
-,,y,script,./test.sh -f tsim/stream/schedSnode.sim
-,,y,script,./test.sh -f tsim/stream/windowClose.sim
-,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
-,,y,script,./test.sh -f tsim/stream/sliding.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
-,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
-,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
-,,y,script,./test.sh -f tsim/stream/deleteSession.sim
-,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalLinear.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
+,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext.sim
+,,y,script,./test.sh -f tsim/stream/fillIntervalRange.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalValue.sim
+,,y,script,./test.sh -f tsim/stream/ignoreCheckUpdate.sim
+,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
+,,y,script,./test.sh -f tsim/stream/partitionby1.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
+,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
+,,y,script,./test.sh -f tsim/stream/partitionby.sim
+,,y,script,./test.sh -f tsim/stream/schedSnode.sim
+,,y,script,./test.sh -f tsim/stream/session0.sim
+,,y,script,./test.sh -f tsim/stream/session1.sim
+,,y,script,./test.sh -f tsim/stream/sliding.sim
+,,y,script,./test.sh -f tsim/stream/state0.sim
+,,y,script,./test.sh -f tsim/stream/state1.sim
+,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
+,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag1.sim
+,,y,script,./test.sh -f tsim/stream/udTableAndTag2.sim
+,,y,script,./test.sh -f tsim/stream/windowClose.sim
,,y,script,./test.sh -f tsim/trans/lossdata1.sim
,,y,script,./test.sh -f tsim/tmq/basic1.sim
,,y,script,./test.sh -f tsim/tmq/basic2.sim
diff --git a/tests/script/win-test-file b/tests/script/win-test-file
index 2d5a1b3108efb2b884bf7664ff61557f09d2bca5..e8e71f8fdc784a7a812d9ed32cca93250792acb3 100644
--- a/tests/script/win-test-file
+++ b/tests/script/win-test-file
@@ -213,38 +213,47 @@ rem ./test.sh -f tsim/db/alter_replica_13.sim
./test.sh -f tsim/stream/basic0.sim -g
./test.sh -f tsim/stream/basic1.sim
./test.sh -f tsim/stream/basic2.sim
+./test.sh -f tsim/stream/basic3.sim
+./test.sh -f tsim/stream/basic4.sim
+./test.sh -f tsim/stream/checkStreamSTable1.sim
+./test.sh -f tsim/stream/checkStreamSTable.sim
+./test.sh -f tsim/stream/deleteInterval.sim
+./test.sh -f tsim/stream/deleteSession.sim
+./test.sh -f tsim/stream/deleteState.sim
+./test.sh -f tsim/stream/distributeInterval0.sim
+./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
+./test.sh -f tsim/stream/distributeSession0.sim
./test.sh -f tsim/stream/drop_stream.sim
./test.sh -f tsim/stream/fillHistoryBasic1.sim
./test.sh -f tsim/stream/fillHistoryBasic2.sim
./test.sh -f tsim/stream/fillHistoryBasic3.sim
-./test.sh -f tsim/stream/distributeInterval0.sim
-./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
-./test.sh -f tsim/stream/distributeSession0.sim
-./test.sh -f tsim/stream/session0.sim
-./test.sh -f tsim/stream/session1.sim
-./test.sh -f tsim/stream/state0.sim
-./test.sh -f tsim/stream/triggerInterval0.sim
-./test.sh -f tsim/stream/triggerSession0.sim
-./test.sh -f tsim/stream/partitionby.sim
-./test.sh -f tsim/stream/partitionby1.sim
-./test.sh -f tsim/stream/schedSnode.sim
-./test.sh -f tsim/stream/windowClose.sim
-./test.sh -f tsim/stream/ignoreExpiredData.sim
-./test.sh -f tsim/stream/sliding.sim
-./test.sh -f tsim/stream/partitionbyColumnInterval.sim
-./test.sh -f tsim/stream/partitionbyColumnSession.sim
-./test.sh -f tsim/stream/partitionbyColumnState.sim
-./test.sh -f tsim/stream/deleteInterval.sim
-./test.sh -f tsim/stream/deleteSession.sim
-./test.sh -f tsim/stream/deleteState.sim
./test.sh -f tsim/stream/fillIntervalDelete0.sim
./test.sh -f tsim/stream/fillIntervalDelete1.sim
./test.sh -f tsim/stream/fillIntervalLinear.sim
./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
+./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
./test.sh -f tsim/stream/fillIntervalPrevNext.sim
+./test.sh -f tsim/stream/fillIntervalRange.sim
./test.sh -f tsim/stream/fillIntervalValue.sim
+./test.sh -f tsim/stream/ignoreCheckUpdate.sim
+./test.sh -f tsim/stream/ignoreExpiredData.sim
+./test.sh -f tsim/stream/partitionby1.sim
+./test.sh -f tsim/stream/partitionbyColumnInterval.sim
+./test.sh -f tsim/stream/partitionbyColumnSession.sim
+./test.sh -f tsim/stream/partitionbyColumnState.sim
+./test.sh -f tsim/stream/partitionby.sim
+./test.sh -f tsim/stream/schedSnode.sim
+./test.sh -f tsim/stream/session0.sim
+./test.sh -f tsim/stream/session1.sim
+./test.sh -f tsim/stream/sliding.sim
+./test.sh -f tsim/stream/state0.sim
+./test.sh -f tsim/stream/state1.sim
+./test.sh -f tsim/stream/triggerInterval0.sim
+./test.sh -f tsim/stream/triggerSession0.sim
./test.sh -f tsim/stream/udTableAndTag0.sim
./test.sh -f tsim/stream/udTableAndTag1.sim
+./test.sh -f tsim/stream/udTableAndTag2.sim
+./test.sh -f tsim/stream/windowClose.sim
./test.sh -f tsim/trans/lossdata1.sim
./test.sh -f tsim/trans/create_db.sim
./test.sh -f tsim/tmq/basic1.sim