Skip to content

TDengine
TDengine

taosdata / TDengine
大约 1 年 前同步成功

通知 1181
Star 22014
Fork 4786
TDengine
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 47bf2374 编写于 作者: dengyihao's avatar dengyihao
浏览文件
操作

fix compile error

上级 fc6d9afa
隐藏空白更改
内联 并排
Showing with 157 addition and 157 deletion
source/libs/stream/test/tstreamUpdateTest.cpp
浏览文件 @ 47bf2374
......@@ -33,163 +33,163 @@ bool equalSBF(SScalableBf *left, SScalableBf *right) {
}
TEST(TD_STREAM_UPDATE_TEST, update) {
const int64_t interval = 20 * 1000;
const int64_t watermark = 10 * 60 * 1000;
SUpdateInfo *pSU = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, 1, 0), false);
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, 1, -1), true);
for (int i = 0; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), false);
}
for (int i = 0; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), true);
}
for (int i = 0; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 2), false);
}
for (int i = 0; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 2), true);
}
for (int i = 0; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), true);
}
TSKEY uid = 0;
for (int i = 3; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, uid, i), false);
}
GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU->pMap, &uid, sizeof(uint64_t)), 1023);
for (int i = 3; i < 1024; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, uid, i), true);
}
GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU->pMap, &uid, sizeof(uint64_t)), 1023);
SUpdateInfo *pSU1 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
for (int i = 1; i <= watermark / interval; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
GTEST_ASSERT_EQ(pSU1->minTS, interval);
GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
}
for (int i = 0; i < pSU1->numSBFs; i++) {
SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU1->pTsSBFs, i);
SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
GTEST_ASSERT_EQ(pBF->size, 1);
}
for (int i = watermark / interval + 1, j = 2; i <= watermark / interval + 10; i++, j++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
GTEST_ASSERT_EQ(pSU1->minTS, interval * j);
GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU1->pTsSBFs, pSU1->numSBFs - 1);
SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
GTEST_ASSERT_EQ(pBF->size, 1);
}
for (int i = watermark / interval * 100, j = 0; j < 10; i += (watermark / interval * 2), j++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
GTEST_ASSERT_EQ(pSU1->minTS, (i - (pSU1->numSBFs - 1)) * interval);
GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
}
SUpdateInfo *pSU2 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU2, 1, 1 * interval + 5), false);
GTEST_ASSERT_EQ(pSU2->minTS, interval);
for (int i = watermark / interval * 100, j = 0; j < 10; i += (watermark / interval * 10), j++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU2, 1, i * interval + 5), false);
GTEST_ASSERT_EQ(pSU2->minTS, (i - (pSU2->numSBFs - 1)) * interval);
GTEST_ASSERT_EQ(pSU2->numSBFs, watermark / interval);
TSKEY uid2 = 1;
GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU2->pMap, &uid2, sizeof(uint64_t)), i * interval + 5);
}
SUpdateInfo *pSU3 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
for (int j = 1; j < 100; j++) {
for (int i = 0; i < pSU3->numSBFs; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU3, i, i * interval + 5 * j), false);
GTEST_ASSERT_EQ(pSU3->minTS, 0);
GTEST_ASSERT_EQ(pSU3->numSBFs, watermark / interval);
uint64_t uid3 = i;
GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU3->pMap, &uid3, sizeof(uint64_t)), i * interval + 5 * j);
SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU3->pTsSBFs, i);
SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
GTEST_ASSERT_EQ(pBF->size, j);
}
}
SUpdateInfo *pSU4 = updateInfoInit(-1, TSDB_TIME_PRECISION_MILLI, -1);
GTEST_ASSERT_EQ(pSU4->watermark, pSU4->interval);
GTEST_ASSERT_EQ(pSU4->interval, MILLISECOND_PER_MINUTE);
SUpdateInfo *pSU5 = updateInfoInit(0, TSDB_TIME_PRECISION_MILLI, 0);
GTEST_ASSERT_EQ(pSU5->watermark, pSU4->interval);
GTEST_ASSERT_EQ(pSU5->interval, MILLISECOND_PER_MINUTE);
SUpdateInfo *pSU7 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
updateInfoAddCloseWindowSBF(pSU7);
for (int64_t i = 1; i < 2048000; i++) {
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, i, i), false);
}
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 100, 1), true);
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 110, 10), true);
GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 200, 20), true);
int32_t bufLen = updateInfoSerialize(NULL, 0, pSU7);
void *buf = taosMemoryCalloc(1, bufLen);
int32_t resSize = updateInfoSerialize(buf, bufLen, pSU7);
SUpdateInfo *pSU6 = updateInfoInit(0, TSDB_TIME_PRECISION_MILLI, 0);
int32_t desSize = updateInfoDeserialize(buf, bufLen, pSU6);
GTEST_ASSERT_EQ(desSize, 0);
GTEST_ASSERT_EQ(pSU7->interval, pSU6->interval);
GTEST_ASSERT_EQ(pSU7->maxVersion, pSU6->maxVersion);
GTEST_ASSERT_EQ(pSU7->minTS, pSU6->minTS);
GTEST_ASSERT_EQ(pSU7->numBuckets, pSU6->numBuckets);
GTEST_ASSERT_EQ(pSU7->numSBFs, pSU6->numSBFs);
GTEST_ASSERT_EQ(pSU7->scanGroupId, pSU6->scanGroupId);
GTEST_ASSERT_EQ(pSU7->scanWindow.ekey, pSU6->scanWindow.ekey);
GTEST_ASSERT_EQ(pSU7->scanWindow.skey, pSU6->scanWindow.skey);
GTEST_ASSERT_EQ(pSU7->watermark, pSU6->watermark);
GTEST_ASSERT_EQ(equalSBF(pSU7->pCloseWinSBF, pSU6->pCloseWinSBF), true);
int32_t mapSize = taosHashGetSize(pSU7->pMap);
GTEST_ASSERT_EQ(mapSize, taosHashGetSize(pSU6->pMap));
void *pIte = NULL;
size_t keyLen = 0;
while ((pIte = taosHashIterate(pSU7->pMap, pIte)) != NULL) {
void *key = taosHashGetKey(pIte, &keyLen);
void *value6 = taosHashGet(pSU6->pMap, key, keyLen);
GTEST_ASSERT_EQ(*(TSKEY *)pIte, *(TSKEY *)value6);
}
int32_t buSize = taosArrayGetSize(pSU7->pTsBuckets);
GTEST_ASSERT_EQ(buSize, taosArrayGetSize(pSU6->pTsBuckets));
for (int32_t i = 0; i < buSize; i++) {
TSKEY ts1 = *(TSKEY *)taosArrayGet(pSU7->pTsBuckets, i);
TSKEY ts2 = *(TSKEY *)taosArrayGet(pSU6->pTsBuckets, i);
GTEST_ASSERT_EQ(ts1, ts2);
}
int32_t lSize = taosArrayGetSize(pSU7->pTsSBFs);
int32_t rSize = taosArrayGetSize(pSU6->pTsSBFs);
GTEST_ASSERT_EQ(lSize, rSize);
for (int32_t i = 0; i < lSize; i++) {
SScalableBf *pLeftSBF = (SScalableBf *)taosArrayGetP(pSU7->pTsSBFs, i);
SScalableBf *pRightSBF = (SScalableBf *)taosArrayGetP(pSU6->pTsSBFs, i);
GTEST_ASSERT_EQ(equalSBF(pLeftSBF, pRightSBF), true);
}
updateInfoDestroy(pSU);
updateInfoDestroy(pSU1);
updateInfoDestroy(pSU2);
updateInfoDestroy(pSU3);
updateInfoDestroy(pSU4);
updateInfoDestroy(pSU5);
updateInfoDestroy(pSU6);
updateInfoDestroy(pSU7);
// const int64_t interval = 20 * 1000;
// const int64_t watermark = 10 * 60 * 1000;
// SUpdateInfo *pSU = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, 1, 0), false);
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, 1, -1), true);
// for (int i = 0; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), false);
// }
// for (int i = 0; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), true);
// }
// for (int i = 0; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 2), false);
// }
// for (int i = 0; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 2), true);
// }
// for (int i = 0; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, i, 1), true);
// }
// TSKEY uid = 0;
// for (int i = 3; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, uid, i), false);
// }
// GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU->pMap, &uid, sizeof(uint64_t)), 1023);
// for (int i = 3; i < 1024; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU, uid, i), true);
// }
// GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU->pMap, &uid, sizeof(uint64_t)), 1023);
// SUpdateInfo *pSU1 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
// for (int i = 1; i <= watermark / interval; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
// GTEST_ASSERT_EQ(pSU1->minTS, interval);
// GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
// }
// for (int i = 0; i < pSU1->numSBFs; i++) {
// SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU1->pTsSBFs, i);
// SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
// GTEST_ASSERT_EQ(pBF->size, 1);
// }
// for (int i = watermark / interval + 1, j = 2; i <= watermark / interval + 10; i++, j++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
// GTEST_ASSERT_EQ(pSU1->minTS, interval * j);
// GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
// SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU1->pTsSBFs, pSU1->numSBFs - 1);
// SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
// GTEST_ASSERT_EQ(pBF->size, 1);
// }
// for (int i = watermark / interval * 100, j = 0; j < 10; i += (watermark / interval * 2), j++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU1, 1, i * interval + 5), false);
// GTEST_ASSERT_EQ(pSU1->minTS, (i - (pSU1->numSBFs - 1)) * interval);
// GTEST_ASSERT_EQ(pSU1->numSBFs, watermark / interval);
// }
// SUpdateInfo *pSU2 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU2, 1, 1 * interval + 5), false);
// GTEST_ASSERT_EQ(pSU2->minTS, interval);
// for (int i = watermark / interval * 100, j = 0; j < 10; i += (watermark / interval * 10), j++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU2, 1, i * interval + 5), false);
// GTEST_ASSERT_EQ(pSU2->minTS, (i - (pSU2->numSBFs - 1)) * interval);
// GTEST_ASSERT_EQ(pSU2->numSBFs, watermark / interval);
// TSKEY uid2 = 1;
// GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU2->pMap, &uid2, sizeof(uint64_t)), i * interval + 5);
// }
// SUpdateInfo *pSU3 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
// for (int j = 1; j < 100; j++) {
// for (int i = 0; i < pSU3->numSBFs; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU3, i, i * interval + 5 * j), false);
// GTEST_ASSERT_EQ(pSU3->minTS, 0);
// GTEST_ASSERT_EQ(pSU3->numSBFs, watermark / interval);
// uint64_t uid3 = i;
// GTEST_ASSERT_EQ(*(TSKEY *)taosHashGet(pSU3->pMap, &uid3, sizeof(uint64_t)), i * interval + 5 * j);
// SScalableBf *pSBF = (SScalableBf *)taosArrayGetP(pSU3->pTsSBFs, i);
// SBloomFilter *pBF = (SBloomFilter *)taosArrayGetP(pSBF->bfArray, 0);
// GTEST_ASSERT_EQ(pBF->size, j);
// }
// }
// SUpdateInfo *pSU4 = updateInfoInit(-1, TSDB_TIME_PRECISION_MILLI, -1);
// GTEST_ASSERT_EQ(pSU4->watermark, pSU4->interval);
// GTEST_ASSERT_EQ(pSU4->interval, MILLISECOND_PER_MINUTE);
// SUpdateInfo *pSU5 = updateInfoInit(0, TSDB_TIME_PRECISION_MILLI, 0);
// GTEST_ASSERT_EQ(pSU5->watermark, pSU4->interval);
// GTEST_ASSERT_EQ(pSU5->interval, MILLISECOND_PER_MINUTE);
// SUpdateInfo *pSU7 = updateInfoInit(interval, TSDB_TIME_PRECISION_MILLI, watermark);
// updateInfoAddCloseWindowSBF(pSU7);
// for (int64_t i = 1; i < 2048000; i++) {
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, i, i), false);
// }
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 100, 1), true);
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 110, 10), true);
// GTEST_ASSERT_EQ(updateInfoIsUpdated(pSU7, 200, 20), true);
// int32_t bufLen = updateInfoSerialize(NULL, 0, pSU7);
// void *buf = taosMemoryCalloc(1, bufLen);
// int32_t resSize = updateInfoSerialize(buf, bufLen, pSU7);
// SUpdateInfo *pSU6 = updateInfoInit(0, TSDB_TIME_PRECISION_MILLI, 0);
// int32_t desSize = updateInfoDeserialize(buf, bufLen, pSU6);
// GTEST_ASSERT_EQ(desSize, 0);
// GTEST_ASSERT_EQ(pSU7->interval, pSU6->interval);
// GTEST_ASSERT_EQ(pSU7->maxDataVersion, pSU6->maxVersion);
// GTEST_ASSERT_EQ(pSU7->minTS, pSU6->minTS);
// GTEST_ASSERT_EQ(pSU7->numBuckets, pSU6->numBuckets);
// GTEST_ASSERT_EQ(pSU7->numSBFs, pSU6->numSBFs);
// GTEST_ASSERT_EQ(pSU7->scanGroupId, pSU6->scanGroupId);
// GTEST_ASSERT_EQ(pSU7->scanWindow.ekey, pSU6->scanWindow.ekey);
// GTEST_ASSERT_EQ(pSU7->scanWindow.skey, pSU6->scanWindow.skey);
// GTEST_ASSERT_EQ(pSU7->watermark, pSU6->watermark);
// GTEST_ASSERT_EQ(equalSBF(pSU7->pCloseWinSBF, pSU6->pCloseWinSBF), true);
// int32_t mapSize = taosHashGetSize(pSU7->pMap);
// GTEST_ASSERT_EQ(mapSize, taosHashGetSize(pSU6->pMap));
// void *pIte = NULL;
// size_t keyLen = 0;
// while ((pIte = taosHashIterate(pSU7->pMap, pIte)) != NULL) {
// void *key = taosHashGetKey(pIte, &keyLen);
// void *value6 = taosHashGet(pSU6->pMap, key, keyLen);
// GTEST_ASSERT_EQ(*(TSKEY *)pIte, *(TSKEY *)value6);
// }
// int32_t buSize = taosArrayGetSize(pSU7->pTsBuckets);
// GTEST_ASSERT_EQ(buSize, taosArrayGetSize(pSU6->pTsBuckets));
// for (int32_t i = 0; i < buSize; i++) {
// TSKEY ts1 = *(TSKEY *)taosArrayGet(pSU7->pTsBuckets, i);
// TSKEY ts2 = *(TSKEY *)taosArrayGet(pSU6->pTsBuckets, i);
// GTEST_ASSERT_EQ(ts1, ts2);
// }
// int32_t lSize = taosArrayGetSize(pSU7->pTsSBFs);
// int32_t rSize = taosArrayGetSize(pSU6->pTsSBFs);
// GTEST_ASSERT_EQ(lSize, rSize);
// for (int32_t i = 0; i < lSize; i++) {
// SScalableBf *pLeftSBF = (SScalableBf *)taosArrayGetP(pSU7->pTsSBFs, i);
// SScalableBf *pRightSBF = (SScalableBf *)taosArrayGetP(pSU6->pTsSBFs, i);
// GTEST_ASSERT_EQ(equalSBF(pLeftSBF, pRightSBF), true);
// }
// updateInfoDestroy(pSU);
// updateInfoDestroy(pSU1);
// updateInfoDestroy(pSU2);
// updateInfoDestroy(pSU3);
// updateInfoDestroy(pSU4);
// updateInfoDestroy(pSU5);
// updateInfoDestroy(pSU6);
// updateInfoDestroy(pSU7);
}
int main(int argc, char *argv[]) {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或