xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

check update data

This commit is contained in:
liuyao 2023-07-31 17:04:41 +08:00
parent 64d63b0584
commit 32b59af4b1
6 changed files with 13 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/executor/storageapi.h
View File

@ -368,6 +368,8 @@ typedef struct SStateStore {
bool (*updateInfoIsUpdated)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts);
bool (*updateInfoIsTableInserted)(SUpdateInfo* pInfo, int64_t tbUid);
void (*updateInfoDestroy)(SUpdateInfo* pInfo);
void (*windowSBfDelete)(SUpdateInfo *pInfo, uint64_t count);
void (*windowSBfAdd)(SUpdateInfo *pInfo, uint64_t count);
SUpdateInfo* (*updateInfoInitP)(SInterval* pInterval, int64_t watermark);
void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo);

2
include/libs/stream/tstreamUpdate.h
View File

@ -53,6 +53,8 @@ void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo);
void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo);
int32_t updateInfoSerialize(void *buf, int32_t bufLen, const SUpdateInfo *pInfo);
int32_t updateInfoDeserialize(void *buf, int32_t bufLen, SUpdateInfo *pInfo);
void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count);
void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count);
#ifdef __cplusplus
}

2
source/dnode/snode/src/snodeInitApi.c
View File

@ -78,6 +78,8 @@ void initStateStoreAPI(SStateStore* pStore) {
pStore->updateInfoIsUpdated = updateInfoIsUpdated;
pStore->updateInfoIsTableInserted = updateInfoIsTableInserted;
pStore->updateInfoDestroy = updateInfoDestroy;
pStore->windowSBfDelete = windowSBfDelete;
pStore->windowSBfAdd = windowSBfAdd;
pStore->updateInfoInitP = updateInfoInitP;
pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF;

2
source/dnode/vnode/src/vnd/vnodeInitApi.c
View File

@ -180,6 +180,8 @@ void initStateStoreAPI(SStateStore* pStore) {
pStore->updateInfoIsUpdated = updateInfoIsUpdated;
pStore->updateInfoIsTableInserted = updateInfoIsTableInserted;
pStore->updateInfoDestroy = updateInfoDestroy;
pStore->windowSBfDelete = windowSBfDelete;
pStore->windowSBfAdd = windowSBfAdd;
pStore->updateInfoInitP = updateInfoInitP;
pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF;

4
source/libs/executor/src/scanoperator.c
View File

@ -2411,7 +2411,9 @@ void streamScanReloadState(SOperatorInfo* pOperator) {
pInfo->stateStore.updateInfoDestroy(pInfo->pUpdateInfo);
pInfo->pUpdateInfo = pUpInfo;
} else {
pInfo->pUpdateInfo->minTS = TMAX(pInfo->pUpdateInfo->minTS, pUpInfo->minTS);
pInfo->stateStore.windowSBfDelete(pInfo->pUpdateInfo, 1);
pInfo->stateStore.windowSBfAdd(pInfo->pUpdateInfo, 1);
ASSERT(pInfo->pUpdateInfo->minTS > pUpInfo->minTS);
pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pUpInfo->maxDataVersion);
SHashObj* curMap = pInfo->pUpdateInfo->pMap;
void *pIte = taosHashIterate(curMap, NULL);

4
source/libs/stream/src/streamUpdate.c
View File

@ -33,7 +33,7 @@
static int64_t adjustExpEntries(int64_t entries) { return TMIN(DEFAULT_EXPECTED_ENTRIES, entries); }
static void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) {
void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) {
if (pInfo->numSBFs < count) {
count = pInfo->numSBFs;
}
@ -49,7 +49,7 @@ static void clearItemHelper(void *p) {
tScalableBfDestroy(*pBf);
}
static void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) {
void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) {
if (count < pInfo->numSBFs) {
for (uint64_t i = 0; i < count; ++i) {
SScalableBf *pTsSBFs = taosArrayGetP(pInfo->pTsSBFs, 0);