Elgato_dark/TDengine
1
0
Fork 0
mirror of https://github.com/taosdata/TDengine synced 2026-05-24 10:09:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4
TDengine/source/libs/new-stream/test/dataSinkTest.cpp
facetosea 7af4c6c77b test: data sink
2025-06-05 19:32:32 +08:00

786 lines
27 KiB
C++
Raw Blame History

/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <stdint.h>
#include "dataSink.h"
#include "tdatablock.h"
const int64_t baseTestTime1 = 1745142096000;
const int64_t baseTestTime2 = 1745142097000;
SSDataBlock* createTestBlock(int64_t basetime, int64_t timeOffset) {
SSDataBlock* b = NULL;
int32_t code = createDataBlock(&b);
int64_t timeStart = basetime + timeOffset;
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, 8, 1);
blockDataAppendColInfo(b, &infoData);
SColumnInfoData infoData1 = createColumnInfoData(TSDB_DATA_TYPE_BINARY, 40, 2);
blockDataAppendColInfo(b, &infoData1);
blockDataEnsureCapacity(b, 100);
const char* str = "the value of: %d";
char buf[128] = {0};
char varbuf[128] = {0};
int64_t ts = basetime;
for (int32_t i = 0; i < 100; ++i) {
SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 0);
SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 1);
ts = timeStart + i;
if (i & 0x01) {
int32_t len = sprintf(buf, str, timeOffset + i);
STR_TO_VARSTR(varbuf, buf)
colDataSetVal(p0, i, (const char*)&ts, false);
colDataSetVal(p1, i, (const char*)varbuf, false);
memset(varbuf, 0, sizeof(varbuf));
memset(buf, 0, sizeof(buf));
} else {
colDataSetVal(p0, i, (const char*)&ts, false);
colDataSetVal(p1, i, (const char*)varbuf, true);
}
b->info.rows++;
}
SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 0);
SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 1);
printf("binary column length:%d\n", *(int32_t*)p1->pData);
char* pData = colDataGetData(p1, 2);
printf("the second row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
pData = colDataGetData(p1, 3);
printf("the third row: %s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
return b;
}
bool compareBlock(SSDataBlock* b1, SSDataBlock* b2) {
if (b1->info.rows != b2->info.rows) {
return false;
}
for (int32_t i = 0; i < b1->info.rows; ++i) {
SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b1->pDataBlock, 0);
SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b2->pDataBlock, 0);
if (*(int32_t*)colDataGetData(p0, i) != *(int32_t*)colDataGetData(p1, i)) {
return false;
}
}
SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b1->pDataBlock, 1);
SColumnInfoData* p2 = (SColumnInfoData*)taosArrayGet(b2->pDataBlock, 1);
char* pData = colDataGetData(p1, 3);
printf("b1 the third row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
pData = colDataGetData(p1, 3);
printf("b2 the third row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
return true;
}
bool compareBlockRow(SSDataBlock* b1, SSDataBlock* b2, int32_t row1, int32_t row2) {
for (int32_t i = 0; i < b1->pDataBlock->size; ++i) {
SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b1->pDataBlock, i);
SColumnInfoData* p2 = (SColumnInfoData*)taosArrayGet(b2->pDataBlock, i);
if (i == 0) {
if (*(int64_t*)colDataGetData(p1, row1) != *(int64_t*)colDataGetData(p2, row2)) {
return false;
}
continue;
} else {
if (colDataIsNull(p1, b1->info.rows, row1, NULL) != colDataIsNull(p2, b2->info.rows, row2, NULL)) {
return false;
}
if (colDataIsNull(p1, b1->info.rows, row1, NULL) == true) {
continue;
}
if (IS_VAR_DATA_TYPE(p1->info.type)) {
char* pData = colDataGetData(p1, row1);
char* pData2 = colDataGetData(p2, row2);
if (varDataLen(pData) != varDataLen(pData2)) {
return false;
}
if (memcmp(varDataVal(pData), varDataVal(pData2), varDataLen(pData)) != 0) {
return false;
}
} else {
if (*(int32_t*)colDataGetData(p1, row1) != *(int32_t*)colDataGetData(p2, row2)) {
return false;
}
}
}
}
return true;
}
TEST(dataSinkTest, fileInit) {
int32_t code = initStreamDataSinkOnce();
ASSERT_EQ(code, 0);
code = initStreamDataSinkOnce();
ASSERT_EQ(code, 0);
destroyDataSinkMgr();
}
TEST(dataSinkTest, test_name) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = 1;
TSKEY wstart = 0;
TSKEY wend = 100;
void* pCache = NULL;
int32_t code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 1);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
}
TEST(dataSinkTest, putStreamDataCacheTest) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_IMMEDIATE;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache = NULL;
// Test invalid parameters
int32_t code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 1);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 1, 0);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 1);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = moveStreamDataCache(pCache, groupID, wstart, wend, pBlock);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = moveStreamDataCache(NULL, groupID, wstart, wend, pBlock);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = getStreamDataCache(pCache, groupID, wend, wstart, NULL);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
// Test valid parameters
code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
void* pIter = NULL;
// Test invalid parameters
code = getStreamDataCache(pCache, groupID, wend, wstart, &pIter);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = getStreamDataCache(NULL, groupID, wstart, wend, &pIter);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = getStreamDataCache(pCache, groupID, wstart, wend, NULL);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
// Test valid parameters
code = getStreamDataCache(pCache, groupID, wstart, wend, &pIter);
ASSERT_EQ(code, 0);
code = getNextStreamDataCache(&pIter, NULL);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
bool equal = compareBlock(pBlock, pBlock1);
ASSERT_EQ(equal, true);
blockDataDestroy(pBlock1);
blockDataDestroy(pBlock);
pBlock = createTestBlock(baseTestTime1, 100);
streamId = 1;
taskId = 1;
groupID = 2;
cleanMode = DATA_CLEAN_IMMEDIATE;
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
pCache = NULL;
code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
pIter = NULL;
code = getStreamDataCache(pCache, groupID, wstart, wend, &pIter);
ASSERT_EQ(code, 0);
pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
equal = compareBlock(pBlock, pBlock1);
ASSERT_EQ(equal, true);
blockDataDestroy(pBlock1);
blockDataDestroy(pBlock);
pBlock = createTestBlock(baseTestTime1, 0);
streamId = 2;
taskId = 1;
groupID = 2;
cleanMode = DATA_CLEAN_IMMEDIATE;
wstart = baseTestTime1 + 0;
wend = baseTestTime1 + 100;
pCache = NULL;
code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
pIter = NULL;
code = getStreamDataCache(pCache, groupID, wstart, wend, &pIter);
ASSERT_EQ(code, 0);
pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
equal = compareBlock(pBlock, pBlock1);
ASSERT_EQ(equal, true);
blockDataDestroy(pBlock1);
destroyDataSinkMgr();
blockDataDestroy(pBlock);
}
TEST(dataSinkTest, getSlidingStreamData) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_EXPIRED;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
// Test invalid parameters, cleanMode is DATA_CLEAN_EXPIRED, cannot call moveStreamDataCache
code = moveStreamDataCache(pCache, groupID, wstart, wend, pBlock);
ASSERT_EQ(code, TSDB_CODE_STREAM_INTERNAL_ERROR);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
blockDataDestroy(pBlock);
pBlock = createTestBlock(baseTestTime1, 100);
cleanMode = DATA_CLEAN_EXPIRED;
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
void* pIter = NULL;
blockDataDestroy(pBlock);
code = getStreamDataCache(pCache, groupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter, nullptr);
int rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
blockDataDestroy(pBlock1);
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 51);
blockDataDestroy(pBlock1);
pBlock = createTestBlock(baseTestTime1, 200);
cleanMode = DATA_CLEAN_EXPIRED;
wstart = baseTestTime1 + 200;
wend = baseTestTime1 + 300;
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
pIter = NULL;
code = getStreamDataCache(pCache, groupID, baseTestTime1 + 150, baseTestTime1 + 249, &pIter);
ASSERT_EQ(code, 0);
pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
ASSERT_EQ(compareBlockRow(pBlock, pBlock1, 0, 0), true);
ASSERT_EQ(compareBlockRow(pBlock, pBlock1, 1, 1), true);
blockDataDestroy(pBlock);
blockDataDestroy(pBlock1);
destroyDataSinkMgr();
}
TEST(dataSinkTest, moveStreamData) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock, nullptr);
int64_t streamId = 3;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_IMMEDIATE;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
code = moveStreamDataCache(pCache, groupID, wstart, wend, pBlock);
ASSERT_EQ(code, 0);
void* pIter = NULL;
code = getStreamDataCache(pCache, groupID, baseTestTime1, baseTestTime1 + 100, &pIter);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter, nullptr);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
ASSERT_EQ(pBlock1, pBlock);
blockDataDestroy(pBlock);
code = getStreamDataCache(pCache, groupID, baseTestTime1, baseTestTime1 + 100, &pIter);
ASSERT_EQ(code, 0);
ASSERT_EQ(pIter, nullptr);
pBlock1 = NULL;
destroyDataSinkMgr();
}
TEST(dataSinkTest, cancelStreamDataCacheIterateTest) {
int64_t streamId = 3;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_IMMEDIATE;
void* pCache = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock1 = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock1, nullptr);
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
code = moveStreamDataCache(pCache, groupID, wstart, wend, pBlock1);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock2 = createTestBlock(baseTestTime1, 100);
ASSERT_NE(pBlock2, nullptr);
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
code = moveStreamDataCache(pCache, groupID, wstart, wend, pBlock2);
ASSERT_EQ(code, 0);
void* pIter = NULL;
code = getStreamDataCache(pCache, groupID, baseTestTime1, baseTestTime1 + 100, &pIter);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter, nullptr);
SSDataBlock* pBlock = NULL;
code = getNextStreamDataCache(&pIter, &pBlock);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock, nullptr);
ASSERT_NE(pIter, nullptr);
ASSERT_EQ(pBlock1, pBlock);
cancelStreamDataCacheIterate(&pIter);
blockDataDestroy(pBlock1); // pBlock1 has moveout, can destroy
// blockDataDestroy(pBlock2); // pBlock2 has not moveout, can not destroy
destroyDataSinkMgr();
}
TEST(dataSinkTest, putStreamDataRows) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_EXPIRED;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 29);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 30, 79);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 80, 99);
ASSERT_EQ(code, 0);
blockDataDestroy(pBlock);
pBlock = createTestBlock(baseTestTime1, 100);
cleanMode = DATA_CLEAN_EXPIRED;
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 49);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 50, 99);
ASSERT_EQ(code, 0);
void* pIter = NULL;
blockDataDestroy(pBlock);
code = getStreamDataCache(pCache, groupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter, nullptr);
int rows = pBlock1->info.rows;
ASSERT_EQ(rows, 30);
blockDataDestroy(pBlock1);
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 20);
blockDataDestroy(pBlock1);
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
blockDataDestroy(pBlock1);
destroyDataSinkMgr();
}
TEST(dataSinkTest, allWriteToFileTest) {
setDataSinkMaxMemSize(0);
SSDataBlock* pBlock11 = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock11, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_EXPIRED;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache1 = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache1);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 0, 29);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 30, 79);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 80, 99);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock21 = createTestBlock(baseTestTime2, 0);
ASSERT_NE(pBlock21, nullptr);
int64_t streamId2 = 1;
int64_t taskId2 = 1;
int64_t groupID2 = 2;
int32_t cleanMode2 = DATA_CLEAN_EXPIRED;
TSKEY wstart2 = baseTestTime2 + 0;
TSKEY wend2 = baseTestTime2 + 100;
void* pCache2 = NULL;
code = initStreamDataCache(streamId2, taskId2, 0, cleanMode2, 0, &pCache2);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 0, 29);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 30, 79);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 80, 99);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock12 = createTestBlock(baseTestTime1, 100);
cleanMode = DATA_CLEAN_EXPIRED;
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock12, 0, 49);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock12, 50, 99);
ASSERT_EQ(code, 0);
void* pIter1 = NULL;
int64_t notExistGroupID = groupID + 100;
code = getStreamDataCache(pCache1, notExistGroupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter1);
ASSERT_EQ(code, 0);
ASSERT_EQ(pIter1, nullptr);
code = getStreamDataCache(pCache1, groupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter1);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter1, nullptr);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter1, nullptr);
int rows = pBlock1->info.rows;
ASSERT_EQ(rows, 30);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 0, 50), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 1, 51), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 2, 52), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 29, 79), true);
blockDataDestroy(pBlock1);
void* pIter2 = NULL;
code = getStreamDataCache(pCache2, groupID2, baseTestTime2 + 50, baseTestTime2 + 150, &pIter2);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter2, nullptr);
SSDataBlock* pBlock2 = NULL;
code = getNextStreamDataCache(&pIter2, &pBlock2);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock2, nullptr);
ASSERT_NE(pIter2, nullptr);
int rows2 = pBlock2->info.rows;
ASSERT_EQ(rows2, 30);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 0, 50), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 1, 51), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 2, 52), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 29, 79), true);
blockDataDestroy(pBlock2);
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 20);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 0, 80), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 1, 81), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 2, 82), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 19, 99), true);
ASSERT_NE(pIter1, nullptr);
blockDataDestroy(pBlock1);
code = getNextStreamDataCache(&pIter2, &pBlock2);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock2, nullptr);
ASSERT_EQ(pIter2, nullptr);
rows = pBlock2->info.rows;
ASSERT_EQ(rows, 20);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 0, 80), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 1, 81), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 2, 82), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 19, 99), true);
blockDataDestroy(pBlock2);
blockDataDestroy(pBlock21);
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter1, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 0, 0), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 1, 1), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 2, 2), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 49, 49), true);
blockDataDestroy(pBlock1);
blockDataDestroy(pBlock11);
blockDataDestroy(pBlock12);
destroyDataSinkMgr();
}
TEST(dataSinkTest, allWriteMultiStreamToFileTest) {
setDataSinkMaxMemSize(0);
SSDataBlock* pBlock11 = createTestBlock(baseTestTime1, 0);
ASSERT_NE(pBlock11, nullptr);
int64_t streamId = 1;
int64_t taskId = 1;
int64_t groupID = 1;
int32_t cleanMode = DATA_CLEAN_EXPIRED;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pCache1 = NULL;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache1);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 0, 29);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 30, 79);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock11, 80, 99);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock21 = createTestBlock(baseTestTime2, 0);
ASSERT_NE(pBlock21, nullptr);
int64_t streamId2 = 2;
int64_t taskId2 = 1;
int64_t groupID2 = 2;
int32_t cleanMode2 = DATA_CLEAN_EXPIRED;
TSKEY wstart2 = baseTestTime2 + 0;
TSKEY wend2 = baseTestTime2 + 100;
void* pCache2 = NULL;
code = initStreamDataCache(streamId2, taskId2, 0, cleanMode2, 0, &pCache2);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 0, 29);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 30, 79);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache2, groupID2, wstart2, wend2, pBlock21, 80, 99);
ASSERT_EQ(code, 0);
SSDataBlock* pBlock12 = createTestBlock(baseTestTime1, 100);
cleanMode = DATA_CLEAN_EXPIRED;
wstart = baseTestTime1 + 100;
wend = baseTestTime1 + 200;
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock12, 0, 49);
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache1, groupID, wstart, wend, pBlock12, 50, 99);
ASSERT_EQ(code, 0);
void* pIter1 = NULL;
int64_t notExistGroupID = groupID + 100;
code = getStreamDataCache(pCache1, notExistGroupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter1);
ASSERT_EQ(code, 0);
ASSERT_EQ(pIter1, nullptr);
code = getStreamDataCache(pCache1, groupID, baseTestTime1 + 50, baseTestTime1 + 150, &pIter1);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter1, nullptr);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter1, nullptr);
int rows = pBlock1->info.rows;
ASSERT_EQ(rows, 30);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 0, 50), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 1, 51), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 2, 52), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 29, 79), true);
blockDataDestroy(pBlock1);
void* pIter2 = NULL;
code = getStreamDataCache(pCache2, groupID2, baseTestTime2 + 50, baseTestTime2 + 150, &pIter2);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter2, nullptr);
SSDataBlock* pBlock2 = NULL;
code = getNextStreamDataCache(&pIter2, &pBlock2);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock2, nullptr);
ASSERT_NE(pIter2, nullptr);
int rows2 = pBlock2->info.rows;
ASSERT_EQ(rows2, 30);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 0, 50), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 1, 51), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 2, 52), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 29, 79), true);
blockDataDestroy(pBlock2);
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 20);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 0, 80), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 1, 81), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 2, 82), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock11, 19, 99), true);
ASSERT_NE(pIter1, nullptr);
blockDataDestroy(pBlock1);
code = getNextStreamDataCache(&pIter2, &pBlock2);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock2, nullptr);
ASSERT_EQ(pIter2, nullptr);
rows = pBlock2->info.rows;
ASSERT_EQ(rows, 20);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 0, 80), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 1, 81), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 2, 82), true);
ASSERT_EQ(compareBlockRow(pBlock2, pBlock21, 19, 99), true);
blockDataDestroy(pBlock2);
blockDataDestroy(pBlock21);
destroyStreamDataCache(pCache2);
code = getNextStreamDataCache(&pIter1, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_NE(pIter1, nullptr);
rows = pBlock1->info.rows;
ASSERT_EQ(rows, 50);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 0, 0), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 1, 1), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 2, 2), true);
ASSERT_EQ(compareBlockRow(pBlock1, pBlock12, 49, 49), true);
blockDataDestroy(pBlock1);
blockDataDestroy(pBlock11);
blockDataDestroy(pBlock12);
destroyStreamDataCache(pCache1);
destroyDataSinkMgr();
}
TEST(dataSinkTest, testWriteFileSize) {
SSDataBlock* pBlock = createTestBlock(baseTestTime1, 0);
setDataSinkMaxMemSize(gMemReservedSize + 1024 * 1024);
int64_t streamId = 1;
void* pCache = NULL;
int64_t taskId = 1;
int32_t cleanMode = DATA_CLEAN_EXPIRED;
int32_t code = initStreamDataCache(streamId, taskId, 0, cleanMode, 0, &pCache);
ASSERT_NE(pBlock, nullptr);
for (int32_t i = 0; i < 100000; i++) {
int64_t groupID = i;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
ASSERT_EQ(code, 0);
code = putStreamDataCache(pCache, groupID, wstart, wend, pBlock, 0, 99);
ASSERT_EQ(code, 0);
}
for (int32_t i = 0; i < 100000; i++) {
int64_t groupID = i;
TSKEY wstart = baseTestTime1 + 0;
TSKEY wend = baseTestTime1 + 100;
void* pIter = NULL;
code = getStreamDataCache(pCache, groupID, wstart, wend, &pIter);
ASSERT_EQ(code, 0);
ASSERT_NE(pIter, nullptr);
SSDataBlock* pBlock1 = NULL;
code = getNextStreamDataCache(&pIter, &pBlock1);
ASSERT_EQ(code, 0);
ASSERT_NE(pBlock1, nullptr);
ASSERT_EQ(pIter, nullptr);
int rows = pBlock1->info.rows;
ASSERT_EQ(rows, 100);
bool equal = compareBlock(pBlock, pBlock1);
ASSERT_EQ(equal, true);
blockDataDestroy(pBlock1);
}
blockDataDestroy(pBlock);
destroyStreamDataCache(pCache);
}
int main(int argc, char** argv) {
taos_init();
::testing::InitGoogleTest(&argc, argv);
if (argc > 1) {
::testing::GTEST_FLAG(filter) = argv[1];
} else {
::testing::GTEST_FLAG(filter) = "*";
}
int ret = RUN_ALL_TESTS();
taos_cleanup();
return ret;
}
Reference in a new issue View git blame Copy permalink