diff --git a/source/libs/stream/src/tstreamUpdate.c b/source/libs/stream/src/tstreamUpdate.c index 1197b6100a72c8438886addd3c2207f0b7961a6e..7921193fa2ca0500afc1467f0e05af267a21aa6c 100644 --- a/source/libs/stream/src/tstreamUpdate.c +++ b/source/libs/stream/src/tstreamUpdate.c @@ -17,21 +17,20 @@ #include "ttime.h" #define DEFAULT_FALSE_POSITIVE 0.01 -#define DEFAULT_BUCKET_SIZE 1024 -#define ROWS_PER_MILLISECOND 1 -#define MAX_NUM_SCALABLE_BF 120 -#define MIN_NUM_SCALABLE_BF 10 -#define DEFAULT_PREADD_BUCKET 1 -#define MAX_INTERVAL MILLISECOND_PER_MINUTE -#define MIN_INTERVAL (MILLISECOND_PER_SECOND * 10) +#define DEFAULT_BUCKET_SIZE 1024 +#define ROWS_PER_MILLISECOND 1 +#define MAX_NUM_SCALABLE_BF 120 +#define MIN_NUM_SCALABLE_BF 10 +#define DEFAULT_PREADD_BUCKET 1 +#define MAX_INTERVAL MILLISECOND_PER_MINUTE +#define MIN_INTERVAL (MILLISECOND_PER_SECOND * 10) static void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) { - if (pInfo->numSBFs < count ) { + if (pInfo->numSBFs < count) { count = pInfo->numSBFs; } for (uint64_t i = 0; i < count; ++i) { - SScalableBf *tsSBF = tScalableBfInit(pInfo->interval * ROWS_PER_MILLISECOND, - DEFAULT_FALSE_POSITIVE); + SScalableBf *tsSBF = tScalableBfInit(pInfo->interval * ROWS_PER_MILLISECOND, DEFAULT_FALSE_POSITIVE); taosArrayPush(pInfo->pTsSBFs, &tsSBF); } } @@ -76,7 +75,7 @@ static int64_t adjustWatermark(int64_t interval, int32_t watermark) { return watermark; } -SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark) { +SUpdateInfo *updateInfoInitP(SInterval *pInterval, int64_t watermark) { return updateInfoInit(pInterval->interval, pInterval->precision, watermark); } @@ -93,7 +92,7 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma uint64_t bfSize = (uint64_t)(pInfo->watermark / pInfo->interval); - pInfo->pTsSBFs = taosArrayInit(bfSize, sizeof(SScalableBf)); + pInfo->pTsSBFs = taosArrayInit(bfSize, sizeof(void *)); if (pInfo->pTsSBFs == NULL) { updateInfoDestroy(pInfo); return NULL; @@ -108,14 +107,14 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma } TSKEY dumy = 0; - for(uint64_t i=0; i < DEFAULT_BUCKET_SIZE; ++i) { + for (uint64_t i = 0; i < DEFAULT_BUCKET_SIZE; ++i) { taosArrayPush(pInfo->pTsBuckets, &dumy); } pInfo->numBuckets = DEFAULT_BUCKET_SIZE; return pInfo; } -static SScalableBf* getSBf(SUpdateInfo *pInfo, TSKEY ts) { +static SScalableBf *getSBf(SUpdateInfo *pInfo, TSKEY ts) { if (ts <= 0) { return NULL; } @@ -131,24 +130,23 @@ static SScalableBf* getSBf(SUpdateInfo *pInfo, TSKEY ts) { } SScalableBf *res = taosArrayGetP(pInfo->pTsSBFs, index); if (res == NULL) { - res = tScalableBfInit(pInfo->interval * ROWS_PER_MILLISECOND, - DEFAULT_FALSE_POSITIVE); + res = tScalableBfInit(pInfo->interval * ROWS_PER_MILLISECOND, DEFAULT_FALSE_POSITIVE); taosArrayPush(pInfo->pTsSBFs, &res); } return res; } bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts) { - int32_t res = TSDB_CODE_FAILED; - uint64_t index = ((uint64_t)tableId) % pInfo->numBuckets; - SScalableBf* pSBf = getSBf(pInfo, ts); + int32_t res = TSDB_CODE_FAILED; + uint64_t index = ((uint64_t)tableId) % pInfo->numBuckets; + SScalableBf *pSBf = getSBf(pInfo, ts); // pSBf may be a null pointer if (pSBf) { res = tScalableBfPut(pSBf, &ts, sizeof(TSKEY)); } TSKEY maxTs = *(TSKEY *)taosArrayGet(pInfo->pTsBuckets, index); - if (maxTs < ts ) { + if (maxTs < ts) { taosArraySet(pInfo->pTsBuckets, index, &ts); return false; } @@ -159,7 +157,7 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts) { return false; } - //check from tsdb api + // check from tsdb api return true; } @@ -174,7 +172,7 @@ void updateInfoDestroy(SUpdateInfo *pInfo) { SScalableBf *pSBF = taosArrayGetP(pInfo->pTsSBFs, i); tScalableBfDestroy(pSBF); } - + taosArrayDestroy(pInfo->pTsSBFs); taosMemoryFree(pInfo); -} \ No newline at end of file +}