TDengine/source/libs/wal/test/walMetaTest.cpp

1500 lines
39 KiB
C++
Raw Permalink Normal View History

2021-12-09 10:32:38 +00:00
#include <gtest/gtest.h>
#include <cstring>
#include <iostream>
#include <queue>
#include <thread>
#include <vector>
#include <atomic>
2021-12-09 10:32:38 +00:00
#include "walInt.h"
#include "tglobal.h"
2021-12-09 10:32:38 +00:00
const char* ranStr = "tvapq02tcp";
const int ranStrLen = strlen(ranStr);
SWalSyncInfo syncMeta = {0};
2021-12-14 06:38:52 +00:00
2021-12-09 10:32:38 +00:00
class WalCleanEnv : public ::testing::Test {
2021-12-16 07:04:13 +00:00
protected:
static void SetUpTestCase() {
2024-08-15 06:07:15 +00:00
int code = walInit(NULL);
2021-12-16 07:04:13 +00:00
ASSERT(code == 0);
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
static void TearDownTestCase() { walCleanUp(); }
void SetUp() override {
taosRemoveDir(pathName);
2022-03-25 16:29:53 +00:00
SWalCfg* pCfg = (SWalCfg*)taosMemoryMalloc(sizeof(SWalCfg));
2021-12-16 07:04:13 +00:00
memset(pCfg, 0, sizeof(SWalCfg));
pCfg->rollPeriod = -1;
pCfg->segSize = -1;
pCfg->retentionPeriod = 0;
pCfg->retentionSize = 0;
pCfg->level = TAOS_WAL_FSYNC;
pWal = walOpen(pathName, pCfg);
2022-03-25 16:29:53 +00:00
taosMemoryFree(pCfg);
2021-12-16 07:04:13 +00:00
ASSERT(pWal != NULL);
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
2021-12-09 10:32:38 +00:00
};
2021-12-11 09:17:21 +00:00
class WalCleanDeleteEnv : public ::testing::Test {
2021-12-16 07:04:13 +00:00
protected:
static void SetUpTestCase() {
2024-08-15 06:07:15 +00:00
int code = walInit(NULL);
2021-12-16 07:04:13 +00:00
ASSERT(code == 0);
}
2021-12-11 09:17:21 +00:00
2021-12-16 07:04:13 +00:00
static void TearDownTestCase() { walCleanUp(); }
void SetUp() override {
taosRemoveDir(pathName);
2022-03-25 16:29:53 +00:00
SWalCfg* pCfg = (SWalCfg*)taosMemoryMalloc(sizeof(SWalCfg));
2021-12-16 07:04:13 +00:00
memset(pCfg, 0, sizeof(SWalCfg));
pCfg->retentionPeriod = 0;
pCfg->retentionSize = 0;
pCfg->level = TAOS_WAL_FSYNC;
pWal = walOpen(pathName, pCfg);
2022-03-25 16:29:53 +00:00
taosMemoryFree(pCfg);
2021-12-16 07:04:13 +00:00
ASSERT(pWal != NULL);
}
2021-12-11 09:17:21 +00:00
2021-12-16 07:04:13 +00:00
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
2021-12-11 09:17:21 +00:00
2021-12-16 07:04:13 +00:00
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
2021-12-11 09:17:21 +00:00
};
2021-12-09 10:32:38 +00:00
class WalKeepEnv : public ::testing::Test {
2021-12-16 07:04:13 +00:00
protected:
static void SetUpTestCase() {
2024-08-15 06:07:15 +00:00
int code = walInit(NULL);
2021-12-16 07:04:13 +00:00
ASSERT(code == 0);
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
static void TearDownTestCase() { walCleanUp(); }
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
void walResetEnv() {
TearDown();
taosRemoveDir(pathName);
SetUp();
}
2021-12-13 05:58:54 +00:00
2021-12-16 07:04:13 +00:00
void SetUp() override {
2022-03-25 16:29:53 +00:00
SWalCfg* pCfg = (SWalCfg*)taosMemoryMalloc(sizeof(SWalCfg));
2021-12-16 07:04:13 +00:00
memset(pCfg, 0, sizeof(SWalCfg));
pCfg->rollPeriod = -1;
pCfg->segSize = -1;
pCfg->retentionPeriod = 0;
pCfg->retentionSize = 0;
pCfg->level = TAOS_WAL_FSYNC;
taosSetSkipKeyCheckMode();
2021-12-16 07:04:13 +00:00
pWal = walOpen(pathName, pCfg);
2022-03-25 16:29:53 +00:00
taosMemoryFree(pCfg);
2021-12-16 07:04:13 +00:00
ASSERT(pWal != NULL);
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
2021-12-09 10:32:38 +00:00
2021-12-16 07:04:13 +00:00
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
2021-12-09 10:32:38 +00:00
};
2022-01-11 07:40:10 +00:00
class WalRetentionEnv : public ::testing::Test {
protected:
static void SetUpTestCase() {
2024-08-15 06:07:15 +00:00
int code = walInit(NULL);
2022-01-11 07:40:10 +00:00
ASSERT(code == 0);
}
static void TearDownTestCase() { walCleanUp(); }
void walResetEnv() {
TearDown();
taosRemoveDir(pathName);
SetUp();
}
void SetUp() override {
2025-12-05 05:26:28 +00:00
SWalCfg cfg = {0};
2022-03-10 09:15:45 +00:00
cfg.rollPeriod = -1;
cfg.segSize = -1;
2024-12-24 09:34:57 +00:00
cfg.committed = -1;
2022-03-10 09:15:45 +00:00
cfg.retentionPeriod = -1;
cfg.retentionSize = 0;
cfg.rollPeriod = 0;
cfg.vgId = 0;
cfg.level = TAOS_WAL_FSYNC;
2022-01-11 07:40:10 +00:00
pWal = walOpen(pathName, &cfg);
ASSERT(pWal != NULL);
}
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
2022-01-11 07:40:10 +00:00
};
2024-12-24 09:34:57 +00:00
class WalSkipLevel : public ::testing::Test {
protected:
static void SetUpTestCase() {
int code = walInit(NULL);
ASSERT(code == 0);
}
static void TearDownTestCase() { walCleanUp(); }
void walResetEnv() {
TearDown();
taosRemoveDir(pathName);
SetUp();
}
void SetUp() override {
SWalCfg cfg = {0};
2024-12-24 09:34:57 +00:00
cfg.rollPeriod = -1;
cfg.segSize = -1;
cfg.committed = -1;
cfg.retentionPeriod = -1;
cfg.retentionSize = 0;
cfg.rollPeriod = 0;
cfg.vgId = 1;
cfg.level = TAOS_WAL_SKIP;
pWal = walOpen(pathName, &cfg);
ASSERT(pWal != NULL);
}
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
};
class WalEncrypted : public ::testing::Test {
protected:
static void SetUpTestCase() {
int code = walInit(NULL);
ASSERT(code == 0);
}
static void TearDownTestCase() { walCleanUp(); }
void walResetEnv() {
TearDown();
taosRemoveDir(pathName);
SetUp();
}
void SetUp() override {
2025-12-19 03:20:59 +00:00
SWalCfg cfg = {0};
2024-12-24 09:34:57 +00:00
cfg.rollPeriod = -1;
cfg.segSize = -1;
cfg.committed = -1;
cfg.retentionPeriod = -1;
cfg.retentionSize = 0;
cfg.rollPeriod = 0;
cfg.vgId = 0;
cfg.level = TAOS_WAL_FSYNC;
2025-12-05 05:26:28 +00:00
cfg.encryptAlgr = 1;
2024-12-24 09:34:57 +00:00
pWal = walOpen(pathName, &cfg);
ASSERT(pWal != NULL);
}
void TearDown() override {
walClose(pWal);
pWal = NULL;
}
SWal* pWal = NULL;
const char* pathName = TD_TMP_DIR_PATH "wal_test";
};
2021-12-09 10:32:38 +00:00
TEST_F(WalCleanEnv, createNew) {
walRollFileInfo(pWal);
ASSERT(pWal->fileInfoSet != NULL);
ASSERT_EQ(pWal->fileInfoSet->size, 1);
2022-07-01 06:39:21 +00:00
SWalFileInfo* pInfo = (SWalFileInfo*)taosArrayGetLast(pWal->fileInfoSet);
2021-12-09 10:32:38 +00:00
ASSERT_EQ(pInfo->firstVer, 0);
ASSERT_EQ(pInfo->lastVer, -1);
ASSERT_EQ(pInfo->closeTs, -1);
ASSERT_EQ(pInfo->fileSize, 0);
}
TEST_F(WalCleanEnv, serialize) {
int code = walRollFileInfo(pWal);
ASSERT(code == 0);
ASSERT(pWal->fileInfoSet != NULL);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
2024-07-22 06:51:22 +00:00
char* ss = NULL;
code = walMetaSerialize(pWal, &ss);
ASSERT(code == 0);
2021-12-09 10:32:38 +00:00
printf("%s\n", ss);
2022-03-25 16:29:53 +00:00
taosMemoryFree(ss);
2021-12-14 10:14:45 +00:00
code = walSaveMeta(pWal);
2021-12-09 10:32:38 +00:00
ASSERT(code == 0);
}
TEST_F(WalCleanEnv, removeOldMeta) {
int code = walRollFileInfo(pWal);
ASSERT(code == 0);
ASSERT(pWal->fileInfoSet != NULL);
2021-12-14 10:14:45 +00:00
code = walSaveMeta(pWal);
2021-12-09 10:32:38 +00:00
ASSERT(code == 0);
code = walRollFileInfo(pWal);
ASSERT(code == 0);
2021-12-14 10:14:45 +00:00
code = walSaveMeta(pWal);
2021-12-09 10:32:38 +00:00
ASSERT(code == 0);
}
2021-12-13 05:58:54 +00:00
TEST_F(WalKeepEnv, readOldMeta) {
walResetEnv();
int code;
2024-07-18 00:04:19 +00:00
syncMeta.isWeek = -1;
syncMeta.seqNum = UINT64_MAX;
syncMeta.term = UINT64_MAX;
2021-12-16 07:04:13 +00:00
for (int i = 0; i < 10; i++) {
code = walAppendLog(pWal, i, i + 1, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-13 05:58:54 +00:00
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
code = walAppendLog(pWal, i + 2, i, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, TSDB_CODE_WAL_INVALID_VER);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
2021-12-13 05:58:54 +00:00
}
2024-07-22 06:51:22 +00:00
char* oldss = NULL;
code = walMetaSerialize(pWal, &oldss);
ASSERT(code == 0);
2021-12-13 05:58:54 +00:00
TearDown();
SetUp();
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.firstVer, 0);
ASSERT_EQ(pWal->vers.lastVer, 9);
2021-12-13 05:58:54 +00:00
2024-07-22 06:51:22 +00:00
char* newss = NULL;
code = walMetaSerialize(pWal, &newss);
ASSERT(code == 0);
2021-12-13 05:58:54 +00:00
2021-12-14 06:38:52 +00:00
int len = strlen(oldss);
2021-12-13 05:58:54 +00:00
ASSERT_EQ(len, strlen(newss));
2021-12-16 07:04:13 +00:00
for (int i = 0; i < len; i++) {
2021-12-13 05:58:54 +00:00
EXPECT_EQ(oldss[i], newss[i]);
}
2022-03-25 16:29:53 +00:00
taosMemoryFree(oldss);
taosMemoryFree(newss);
2021-12-13 05:58:54 +00:00
}
2021-12-11 09:17:21 +00:00
TEST_F(WalCleanEnv, write) {
int code;
2021-12-16 07:04:13 +00:00
for (int i = 0; i < 10; i++) {
code = walAppendLog(pWal, i, i + 1, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-11 09:17:21 +00:00
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
code = walAppendLog(pWal, i + 2, i, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, TSDB_CODE_WAL_INVALID_VER);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
2021-12-09 10:32:38 +00:00
}
2021-12-14 10:14:45 +00:00
code = walSaveMeta(pWal);
2021-12-11 09:17:21 +00:00
ASSERT_EQ(code, 0);
2021-12-09 10:32:38 +00:00
}
2021-12-11 09:17:21 +00:00
TEST_F(WalCleanEnv, rollback) {
2021-12-09 10:32:38 +00:00
int code;
2021-12-16 07:04:13 +00:00
for (int i = 0; i < 10; i++) {
code = walAppendLog(pWal, i, i + 1, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-09 10:32:38 +00:00
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
2021-12-09 10:32:38 +00:00
}
2022-03-09 02:49:42 +00:00
code = walRollback(pWal, 12);
ASSERT_NE(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 9);
code = walRollback(pWal, 9);
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 8);
2021-12-11 09:17:21 +00:00
code = walRollback(pWal, 5);
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, 4);
2021-12-11 09:17:21 +00:00
code = walRollback(pWal, 3);
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, 2);
2021-12-14 10:14:45 +00:00
code = walSaveMeta(pWal);
2021-12-11 09:17:21 +00:00
ASSERT_EQ(code, 0);
}
2022-10-26 10:03:25 +00:00
TEST_F(WalCleanEnv, rollbackMultiFile) {
int code;
for (int i = 0; i < 10; i++) {
code = walAppendLog(pWal, i, i + 1, syncMeta, (void*)ranStr, ranStrLen, NULL);
2022-10-26 10:03:25 +00:00
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, i);
if (i == 5) {
walBeginSnapshot(pWal, i, 0);
walEndSnapshot(pWal, false);
2022-10-26 10:03:25 +00:00
}
}
code = walRollback(pWal, 12);
ASSERT_NE(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 9);
code = walRollback(pWal, 9);
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 8);
code = walRollback(pWal, 6);
2022-10-26 10:03:25 +00:00
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 5);
code = walRollback(pWal, 5);
2024-07-19 09:51:41 +00:00
ASSERT_NE(code, 0);
2022-10-26 10:03:25 +00:00
ASSERT_EQ(pWal->vers.lastVer, 5);
2022-10-26 10:03:25 +00:00
code = walAppendLog(pWal, 6, 6, syncMeta, (void*)ranStr, ranStrLen, NULL);
2022-10-26 10:03:25 +00:00
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, 6);
2022-10-26 10:03:25 +00:00
code = walSaveMeta(pWal);
ASSERT_EQ(code, 0);
}
2021-12-11 09:17:21 +00:00
TEST_F(WalCleanDeleteEnv, roll) {
int code;
int i;
2021-12-16 07:04:13 +00:00
for (i = 0; i < 100; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-11 09:17:21 +00:00
ASSERT_EQ(code, 0);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.lastVer, i);
2021-12-11 09:17:21 +00:00
code = walCommit(pWal, i);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.commitVer, i);
2021-12-11 09:17:21 +00:00
}
walBeginSnapshot(pWal, i - 1, 0);
2021-12-16 07:04:13 +00:00
ASSERT_EQ(pWal->vers.verInSnapshotting, i - 1);
walEndSnapshot(pWal, false);
2021-12-16 07:04:13 +00:00
ASSERT_EQ(pWal->vers.snapshotVer, i - 1);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.verInSnapshotting, -1);
2021-12-11 09:17:21 +00:00
code = walAppendLog(pWal, 5, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-11 09:17:21 +00:00
ASSERT_NE(code, 0);
2021-12-16 07:04:13 +00:00
for (; i < 200; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
2021-12-11 09:17:21 +00:00
ASSERT_EQ(code, 0);
code = walCommit(pWal, i);
2021-12-13 10:33:30 +00:00
ASSERT_EQ(pWal->vers.commitVer, i);
2021-12-11 09:17:21 +00:00
}
code = walBeginSnapshot(pWal, i - 1, 0);
2021-12-13 05:58:54 +00:00
ASSERT_EQ(code, 0);
code = walEndSnapshot(pWal, false);
2021-12-09 10:32:38 +00:00
ASSERT_EQ(code, 0);
}
2021-12-14 06:38:52 +00:00
TEST_F(WalKeepEnv, readHandleRead) {
walResetEnv();
2022-07-07 03:42:45 +00:00
int code;
feat: new stream (#31678) * fix: windows compile issue * test: add vtable cases (#31829) * fix: windows compile issues * test:add test cases * fix: windows compile issue * case: em-4 stream case submit * test: stream4_sub1 found bug2 * test: submit test_scene_meters_bug2.py * add stream parameters example * feat: [TS-6100] Do not translate const value as column. * Feat/ts 6100 3.0 zlv (#31747) * modify asan exampel * modify asan exampel * add example * add example * modify case example --------- Co-authored-by: zelv01 <1101510017@qq.com> * feat(stream): fix memory leak * modify sliding example * test: update test case. * feat(stream): fix conflicts * fix: add offset case 10a 10s 10m 10h 10d * feat(stream): fix conflicts * chore(stream): rename case name #TS-6100 * add case * modify example * fix: windows compile issues * fix: data null check * feat: [TS-6100] Forbid where when using %%trows (#31827) * feat: [TS-6100] Forbid where when using %%trows * test: update cases * feat: [TS-6100] Fix leaks. --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * test: reproduce bugs * test: update test case. * test: update test case. * feat: [TS-6100] Fix leaks. * test: add cases * Feat/ts 6100 3.0.pw10 (#31841) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * test: reproduce bugs * fix: add sliding interval combine case * test: add cases * test: add recalc test. * test: reproduce bugs * case : add vt ts is null check * modify case * bug: submit test_idmp_meters_bug3.py * test: add test for recalc. * test: add cases * fix: error code check * test: add cases * fix(stream): scan wal with schema in that version * add case * test: add cases * test: update test case. * fix: windows compile issues * add case * test: add cases (#31845) * modify case * fix: reset interpPrev * test: add test_idmp_meters bug4 and bug3 * add case * fix(stream): opti wal interface * fix: remove test_idmp_meters_bug5.py * test: add cases * fix(stream): fix ts data fetch for virtual tables * cancel asan case * test: update test case. * test: update test case. * add case * test: add cases * test: add cases * test: add case test_idmp_meters_bug5.py * test: update test case. * fix(stream): tmq error * test: add cases * feat: [TS-6100] Restore deleted code in mndSma.c since they are still in use. * fix(stream): optimize val scan logic * test: add test_recalc_expired_time.py to ci. * test: update test case. * test: update test case. * feat: [TS-6100] Fix fill range check * fix(stream): optimize val scan logic * add case * test: modify for partition by %%1 * test: add fun case stream4_sub7 * fix(stream): optimize val scan logic * add case * feat: [TS-6100] Rename OPTIONS to STREAM_OPTIONS. * test: add test for recalc. * test: use stream_options. * fix: some cases error. * test: remove recalc from ci. * fix: ci case issues (#31880) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): fix compilation error * fix(stream): optimize val scan logic * test:add test cases * test: modify case * fix: external agg error * test(stream): tobacco scene testing #TD-36514 * test: add stream cases (#31885) * fix: windows compile issue * fix: calc timerange * fix: windows compile issue * modify case * fix(stream): compile error * test: remove one debug test case file * test: modify * test: add test cases * test: reproduce bugs * test: reproduce bugs * feat: [TS-6100] Placeholder function should only appera in SELECT and… (#31868) * feat: [TS-6100] Placeholder function should only appera in SELECT and WHERE and FROM. * test: update case --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * add example * add example * modify case example * modify case * test:alter sql * test: add stream5 case * fix(stream): get schema error with version * test: add delete recalc test py. * test: remove bug cases * test: stream5 case test passed * test: add state cases (#31893) * fix(stream): compile error * test: modify case * test: add cases * test: add test. * test: update test case. * chore(test): fix case err * test: update test case. * fix: align data get * fix(stream): fix row index of datablock written into data cache * fix: put align data * test: update test case. * test: add test cases for virtual table * chore(test): fix case err #TD-36514 * add case * test: add test for water mark. * test: add meters bug6 for stream5 * test: add cases (#31903) * test: add test for recalc. * feat: [TS-6100] %%trows can only be used when event type is window close. * test: add precision of database for ms/us/ns * modify case * add case * add case * test: add test to ci. * modify case * fix: ci case issues (#31904) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues * fix: ci case issues * fix: drop dnode issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): ci error * test: update test case. * feat: [TS-6100] Disable some failed UT. * feat: [TS-6100] Fix virtual table * test: add bug 5. * test: add test delete recalc to ci. * test: add bug 6. * test(stream): tobacco scene #TD-36514 * fix: reqCids,reqCols memory leak in SSTriggerRealtimeContext Co-authored-by: Tony Zhang <tonyzhang@taosdata.com> * test: add case stream6 * fix(stream): implement some pending features in trigger task * modify case * modify case * fix: case issues * modify case * test: add recalc for warter mark. * fix(stream): fix count window trigger of virtual tables * fix(stream): memory leak * test: fix run err. * test: add stream6 bug7 * fix: adjust format * test(stream): tobacco scene testing #TD-36514 * test: change bug7 with update window1 and 2 * test: add test bug 7. * case: restore write 3 window * fix: windows compile issue * fix: notify * test: add cases * modify case * test: update test case. * test(stream): toobacco scene testing #TD-36514 --------- Co-authored-by: Simon Guan <slguan@taosdata.com> Co-authored-by: plum-lihui <huili@taosdata.com> Co-authored-by: Alex Duan <417921451@qq.com> Co-authored-by: zelv01 <1101510017@qq.com> Co-authored-by: Jing Sima <simondominic9997@outlook.com> Co-authored-by: xiangyang guo <66111494+happyguoxy@users.noreply.github.com> Co-authored-by: wangmm0220 <wangmm0220@gmail.com> Co-authored-by: Haojun Liao <hjliao@taosdata.com> Co-authored-by: zyyang90 <zyyang@taosdata.com> Co-authored-by: Alex Duan <51781608+DuanKuanJun@users.noreply.github.com> Co-authored-by: facetosea <285808407@qq.com> Co-authored-by: Simon Guan <guanshengliang@qq.com> Co-authored-by: huohong <sallyhuo@taosdata.com> Co-authored-by: Li Hui <52318143+plum-lihui@users.noreply.github.com> Co-authored-by: Jinqing Kuang <kuangjinqingcn@gmail.com> Co-authored-by: xiao-77 <berylbao@taosdata.com> Co-authored-by: Zhixiao Bao <62235797+xiao-77@users.noreply.github.com> Co-authored-by: happyguoxy <happy_guoxy@163.com> Co-authored-by: Tony Zhang <34825804+Tony2h@users.noreply.github.com> Co-authored-by: Tony Zhang <tonyzhang@taosdata.com>
2025-07-16 06:42:16 +00:00
SWalReader* pRead = walOpenReader(pWal, 0);
2021-12-14 06:38:52 +00:00
ASSERT(pRead != NULL);
2021-12-16 07:04:13 +00:00
int i;
for (i = 0; i < 100; i++) {
2021-12-14 06:38:52 +00:00
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2021-12-14 06:38:52 +00:00
ASSERT_EQ(code, 0);
}
2021-12-16 07:04:13 +00:00
for (int i = 0; i < 1000; i++) {
2022-03-09 16:36:30 +00:00
int ver = taosRand() % 100;
2022-07-07 03:42:45 +00:00
code = walReadVer(pRead, ver);
2021-12-14 06:38:52 +00:00
ASSERT_EQ(code, 0);
2021-12-16 07:04:13 +00:00
// printf("rrbody: \n");
// for(int i = 0; i < pRead->pHead->head.len; i++) {
// printf("%d ", pRead->pHead->head.body[i]);
2021-12-14 06:38:52 +00:00
//}
2021-12-16 07:04:13 +00:00
// printf("\n");
2021-12-14 06:38:52 +00:00
ASSERT_EQ(pRead->pHead->head.version, ver);
2021-12-16 07:04:13 +00:00
ASSERT_EQ(pRead->curVersion, ver + 1);
2021-12-14 06:38:52 +00:00
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, ver);
int len = strlen(newStr);
2022-04-24 05:42:54 +00:00
ASSERT_EQ(pRead->pHead->head.bodyLen, len);
2021-12-16 07:04:13 +00:00
for (int j = 0; j < len; j++) {
2021-12-14 06:38:52 +00:00
EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
}
}
2022-07-07 03:42:45 +00:00
walCloseReader(pRead);
2021-12-14 06:38:52 +00:00
}
2022-01-11 07:40:10 +00:00
2024-12-24 09:34:57 +00:00
TEST_F(WalKeepEnv, walLogExist) {
walResetEnv();
int code;
feat: new stream (#31678) * fix: windows compile issue * test: add vtable cases (#31829) * fix: windows compile issues * test:add test cases * fix: windows compile issue * case: em-4 stream case submit * test: stream4_sub1 found bug2 * test: submit test_scene_meters_bug2.py * add stream parameters example * feat: [TS-6100] Do not translate const value as column. * Feat/ts 6100 3.0 zlv (#31747) * modify asan exampel * modify asan exampel * add example * add example * modify case example --------- Co-authored-by: zelv01 <1101510017@qq.com> * feat(stream): fix memory leak * modify sliding example * test: update test case. * feat(stream): fix conflicts * fix: add offset case 10a 10s 10m 10h 10d * feat(stream): fix conflicts * chore(stream): rename case name #TS-6100 * add case * modify example * fix: windows compile issues * fix: data null check * feat: [TS-6100] Forbid where when using %%trows (#31827) * feat: [TS-6100] Forbid where when using %%trows * test: update cases * feat: [TS-6100] Fix leaks. --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * test: reproduce bugs * test: update test case. * test: update test case. * feat: [TS-6100] Fix leaks. * test: add cases * Feat/ts 6100 3.0.pw10 (#31841) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * test: reproduce bugs * fix: add sliding interval combine case * test: add cases * test: add recalc test. * test: reproduce bugs * case : add vt ts is null check * modify case * bug: submit test_idmp_meters_bug3.py * test: add test for recalc. * test: add cases * fix: error code check * test: add cases * fix(stream): scan wal with schema in that version * add case * test: add cases * test: update test case. * fix: windows compile issues * add case * test: add cases (#31845) * modify case * fix: reset interpPrev * test: add test_idmp_meters bug4 and bug3 * add case * fix(stream): opti wal interface * fix: remove test_idmp_meters_bug5.py * test: add cases * fix(stream): fix ts data fetch for virtual tables * cancel asan case * test: update test case. * test: update test case. * add case * test: add cases * test: add cases * test: add case test_idmp_meters_bug5.py * test: update test case. * fix(stream): tmq error * test: add cases * feat: [TS-6100] Restore deleted code in mndSma.c since they are still in use. * fix(stream): optimize val scan logic * test: add test_recalc_expired_time.py to ci. * test: update test case. * test: update test case. * feat: [TS-6100] Fix fill range check * fix(stream): optimize val scan logic * add case * test: modify for partition by %%1 * test: add fun case stream4_sub7 * fix(stream): optimize val scan logic * add case * feat: [TS-6100] Rename OPTIONS to STREAM_OPTIONS. * test: add test for recalc. * test: use stream_options. * fix: some cases error. * test: remove recalc from ci. * fix: ci case issues (#31880) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): fix compilation error * fix(stream): optimize val scan logic * test:add test cases * test: modify case * fix: external agg error * test(stream): tobacco scene testing #TD-36514 * test: add stream cases (#31885) * fix: windows compile issue * fix: calc timerange * fix: windows compile issue * modify case * fix(stream): compile error * test: remove one debug test case file * test: modify * test: add test cases * test: reproduce bugs * test: reproduce bugs * feat: [TS-6100] Placeholder function should only appera in SELECT and… (#31868) * feat: [TS-6100] Placeholder function should only appera in SELECT and WHERE and FROM. * test: update case --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * add example * add example * modify case example * modify case * test:alter sql * test: add stream5 case * fix(stream): get schema error with version * test: add delete recalc test py. * test: remove bug cases * test: stream5 case test passed * test: add state cases (#31893) * fix(stream): compile error * test: modify case * test: add cases * test: add test. * test: update test case. * chore(test): fix case err * test: update test case. * fix: align data get * fix(stream): fix row index of datablock written into data cache * fix: put align data * test: update test case. * test: add test cases for virtual table * chore(test): fix case err #TD-36514 * add case * test: add test for water mark. * test: add meters bug6 for stream5 * test: add cases (#31903) * test: add test for recalc. * feat: [TS-6100] %%trows can only be used when event type is window close. * test: add precision of database for ms/us/ns * modify case * add case * add case * test: add test to ci. * modify case * fix: ci case issues (#31904) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues * fix: ci case issues * fix: drop dnode issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): ci error * test: update test case. * feat: [TS-6100] Disable some failed UT. * feat: [TS-6100] Fix virtual table * test: add bug 5. * test: add test delete recalc to ci. * test: add bug 6. * test(stream): tobacco scene #TD-36514 * fix: reqCids,reqCols memory leak in SSTriggerRealtimeContext Co-authored-by: Tony Zhang <tonyzhang@taosdata.com> * test: add case stream6 * fix(stream): implement some pending features in trigger task * modify case * modify case * fix: case issues * modify case * test: add recalc for warter mark. * fix(stream): fix count window trigger of virtual tables * fix(stream): memory leak * test: fix run err. * test: add stream6 bug7 * fix: adjust format * test(stream): tobacco scene testing #TD-36514 * test: change bug7 with update window1 and 2 * test: add test bug 7. * case: restore write 3 window * fix: windows compile issue * fix: notify * test: add cases * modify case * test: update test case. * test(stream): toobacco scene testing #TD-36514 --------- Co-authored-by: Simon Guan <slguan@taosdata.com> Co-authored-by: plum-lihui <huili@taosdata.com> Co-authored-by: Alex Duan <417921451@qq.com> Co-authored-by: zelv01 <1101510017@qq.com> Co-authored-by: Jing Sima <simondominic9997@outlook.com> Co-authored-by: xiangyang guo <66111494+happyguoxy@users.noreply.github.com> Co-authored-by: wangmm0220 <wangmm0220@gmail.com> Co-authored-by: Haojun Liao <hjliao@taosdata.com> Co-authored-by: zyyang90 <zyyang@taosdata.com> Co-authored-by: Alex Duan <51781608+DuanKuanJun@users.noreply.github.com> Co-authored-by: facetosea <285808407@qq.com> Co-authored-by: Simon Guan <guanshengliang@qq.com> Co-authored-by: huohong <sallyhuo@taosdata.com> Co-authored-by: Li Hui <52318143+plum-lihui@users.noreply.github.com> Co-authored-by: Jinqing Kuang <kuangjinqingcn@gmail.com> Co-authored-by: xiao-77 <berylbao@taosdata.com> Co-authored-by: Zhixiao Bao <62235797+xiao-77@users.noreply.github.com> Co-authored-by: happyguoxy <happy_guoxy@163.com> Co-authored-by: Tony Zhang <34825804+Tony2h@users.noreply.github.com> Co-authored-by: Tony Zhang <tonyzhang@taosdata.com>
2025-07-16 06:42:16 +00:00
SWalReader* pRead = walOpenReader(pWal, 0);
2024-12-24 09:34:57 +00:00
ASSERT(pRead != NULL);
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
ASSERT_EQ(code, 0);
}
walLogExist(pWal, 0);
ASSERT_EQ(code, 0);
walCloseReader(pRead);
}
TEST_F(WalKeepEnv, walScanLogGetLastVerHeadMissMatch) {
walResetEnv();
int code;
do {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, 0);
sprintf(newStr, "%s-%d", ranStr, 0);
int len = strlen(newStr);
code = walAppendLog(pWal, 0, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
} while (0);
int i = 0;
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
int64_t offset = walGetCurFileOffset(pWal);
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
pWal->writeHead.head.version = i;
pWal->writeHead.head.bodyLen = len;
pWal->writeHead.head.msgType = 0;
pWal->writeHead.head.ingestTs = taosGetTimestampUs();
pWal->writeHead.head.syncMeta = syncMeta;
pWal->writeHead.cksumHead = 1;
pWal->writeHead.cksumBody = walCalcBodyCksum(newStr, len);
taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead));
taosWriteFile(pWal->pLogFile, newStr, len);
int64_t lastVer = 0;
code = walScanLogGetLastVer(pWal, 0, &lastVer);
ASSERT_EQ(code, TSDB_CODE_WAL_CHKSUM_MISMATCH);
}
TEST_F(WalKeepEnv, walScanLogGetLastVerBodyMissMatch) {
walResetEnv();
int code;
do {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, 0);
sprintf(newStr, "%s-%d", ranStr, 0);
int len = strlen(newStr);
code = walAppendLog(pWal, 0, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
} while (0);
int i = 0;
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
int64_t offset = walGetCurFileOffset(pWal);
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
pWal->writeHead.head.version = i;
pWal->writeHead.head.bodyLen = len;
pWal->writeHead.head.msgType = 0;
pWal->writeHead.head.ingestTs = taosGetTimestampUs();
pWal->writeHead.head.syncMeta = syncMeta;
pWal->writeHead.cksumHead = walCalcHeadCksum(&pWal->writeHead);
pWal->writeHead.cksumBody = 1;
taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead));
taosWriteFile(pWal->pLogFile, newStr, len);
int64_t lastVer = 0;
code = walScanLogGetLastVer(pWal, 0, &lastVer);
ASSERT_EQ(code, TSDB_CODE_WAL_CHKSUM_MISMATCH);
}
TEST_F(WalKeepEnv, walCheckAndRepairIdxFile) {
walResetEnv();
int code;
do {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, 0);
sprintf(newStr, "%s-%d", ranStr, 0);
int len = strlen(newStr);
code = walAppendLog(pWal, 0, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
} while (0);
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
for (int i = 1; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
pWal->writeHead.head.version = i;
pWal->writeHead.head.bodyLen = len;
pWal->writeHead.head.msgType = 0;
pWal->writeHead.head.ingestTs = taosGetTimestampUs();
pWal->writeHead.head.syncMeta = syncMeta;
pWal->writeHead.cksumHead = walCalcHeadCksum(&pWal->writeHead);
pWal->writeHead.cksumBody = walCalcBodyCksum(newStr, len);
taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead));
taosWriteFile(pWal->pLogFile, newStr, len);
}
pWal->vers.lastVer = 99;
pFileInfo->lastVer = 99;
code = walCheckAndRepairIdx(pWal);
ASSERT_EQ(code, 0);
}
TEST_F(WalKeepEnv, walRestoreFromSnapshot1) {
walResetEnv();
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
ASSERT_EQ(code, 0);
}
code = walRestoreFromSnapshot(pWal, 50);
ASSERT_EQ(code, 0);
}
TEST_F(WalKeepEnv, walRestoreFromSnapshot2) {
walResetEnv();
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
ASSERT_EQ(code, 0);
}
SWalRef* ref = walOpenRef(pWal);
ref->refVer = 10;
code = walRestoreFromSnapshot(pWal, 99);
ASSERT_EQ(code, -1);
}
TEST_F(WalKeepEnv, walRollback) {
walResetEnv();
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-24 09:34:57 +00:00
ASSERT_EQ(code, 0);
}
code = walRollback(pWal, -1);
ASSERT_EQ(code, TSDB_CODE_WAL_INVALID_VER);
pWal->vers.lastVer = 50;
pWal->vers.commitVer = 40;
pWal->vers.snapshotVer = 40;
SWalFileInfo* fileInfo = walGetCurFileInfo(pWal);
code = walRollback(pWal, 48);
ASSERT_EQ(code, 0);
}
2024-12-30 07:48:11 +00:00
TEST_F(WalKeepEnv, walChangeWrite) {
walResetEnv();
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-30 07:48:11 +00:00
ASSERT_EQ(code, 0);
}
code = walChangeWrite(pWal, 50);
ASSERT_EQ(code, 0);
}
2024-12-27 09:43:42 +00:00
TEST_F(WalCleanEnv, walRepairLogFileTs2) {
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-27 09:43:42 +00:00
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-27 09:43:42 +00:00
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (i = 200; i < 300; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2024-12-27 09:43:42 +00:00
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
// Try to step in ts repair logic.
SWalFileInfo* pFileInfo = (SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, 2);
pFileInfo->closeTs = -1;
code = walCheckAndRepairMeta(pWal);
ASSERT_EQ(code, 0);
}
2022-01-11 07:40:10 +00:00
TEST_F(WalRetentionEnv, repairMeta1) {
walResetEnv();
int code;
2022-03-09 02:49:42 +00:00
2022-01-11 07:40:10 +00:00
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2022-01-11 07:40:10 +00:00
ASSERT_EQ(code, 0);
}
TearDown();
2022-03-09 02:49:42 +00:00
// getchar();
2022-01-11 07:40:10 +00:00
char buf[100];
sprintf(buf, "%s/meta-ver%d", pathName, 0);
taosRemoveFile(buf);
2022-01-11 08:55:40 +00:00
sprintf(buf, "%s/meta-ver%d", pathName, 1);
taosRemoveFile(buf);
2022-01-11 07:40:10 +00:00
SetUp();
2022-03-09 02:49:42 +00:00
// getchar();
2022-01-11 07:40:10 +00:00
ASSERT_EQ(pWal->vers.lastVer, 99);
feat: new stream (#31678) * fix: windows compile issue * test: add vtable cases (#31829) * fix: windows compile issues * test:add test cases * fix: windows compile issue * case: em-4 stream case submit * test: stream4_sub1 found bug2 * test: submit test_scene_meters_bug2.py * add stream parameters example * feat: [TS-6100] Do not translate const value as column. * Feat/ts 6100 3.0 zlv (#31747) * modify asan exampel * modify asan exampel * add example * add example * modify case example --------- Co-authored-by: zelv01 <1101510017@qq.com> * feat(stream): fix memory leak * modify sliding example * test: update test case. * feat(stream): fix conflicts * fix: add offset case 10a 10s 10m 10h 10d * feat(stream): fix conflicts * chore(stream): rename case name #TS-6100 * add case * modify example * fix: windows compile issues * fix: data null check * feat: [TS-6100] Forbid where when using %%trows (#31827) * feat: [TS-6100] Forbid where when using %%trows * test: update cases * feat: [TS-6100] Fix leaks. --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * test: reproduce bugs * test: update test case. * test: update test case. * feat: [TS-6100] Fix leaks. * test: add cases * Feat/ts 6100 3.0.pw10 (#31841) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * test: reproduce bugs * fix: add sliding interval combine case * test: add cases * test: add recalc test. * test: reproduce bugs * case : add vt ts is null check * modify case * bug: submit test_idmp_meters_bug3.py * test: add test for recalc. * test: add cases * fix: error code check * test: add cases * fix(stream): scan wal with schema in that version * add case * test: add cases * test: update test case. * fix: windows compile issues * add case * test: add cases (#31845) * modify case * fix: reset interpPrev * test: add test_idmp_meters bug4 and bug3 * add case * fix(stream): opti wal interface * fix: remove test_idmp_meters_bug5.py * test: add cases * fix(stream): fix ts data fetch for virtual tables * cancel asan case * test: update test case. * test: update test case. * add case * test: add cases * test: add cases * test: add case test_idmp_meters_bug5.py * test: update test case. * fix(stream): tmq error * test: add cases * feat: [TS-6100] Restore deleted code in mndSma.c since they are still in use. * fix(stream): optimize val scan logic * test: add test_recalc_expired_time.py to ci. * test: update test case. * test: update test case. * feat: [TS-6100] Fix fill range check * fix(stream): optimize val scan logic * add case * test: modify for partition by %%1 * test: add fun case stream4_sub7 * fix(stream): optimize val scan logic * add case * feat: [TS-6100] Rename OPTIONS to STREAM_OPTIONS. * test: add test for recalc. * test: use stream_options. * fix: some cases error. * test: remove recalc from ci. * fix: ci case issues (#31880) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): fix compilation error * fix(stream): optimize val scan logic * test:add test cases * test: modify case * fix: external agg error * test(stream): tobacco scene testing #TD-36514 * test: add stream cases (#31885) * fix: windows compile issue * fix: calc timerange * fix: windows compile issue * modify case * fix(stream): compile error * test: remove one debug test case file * test: modify * test: add test cases * test: reproduce bugs * test: reproduce bugs * feat: [TS-6100] Placeholder function should only appera in SELECT and… (#31868) * feat: [TS-6100] Placeholder function should only appera in SELECT and WHERE and FROM. * test: update case --------- Co-authored-by: Simon Guan <guanshengliang@qq.com> * add example * add example * modify case example * modify case * test:alter sql * test: add stream5 case * fix(stream): get schema error with version * test: add delete recalc test py. * test: remove bug cases * test: stream5 case test passed * test: add state cases (#31893) * fix(stream): compile error * test: modify case * test: add cases * test: add test. * test: update test case. * chore(test): fix case err * test: update test case. * fix: align data get * fix(stream): fix row index of datablock written into data cache * fix: put align data * test: update test case. * test: add test cases for virtual table * chore(test): fix case err #TD-36514 * add case * test: add test for water mark. * test: add meters bug6 for stream5 * test: add cases (#31903) * test: add test for recalc. * feat: [TS-6100] %%trows can only be used when event type is window close. * test: add precision of database for ms/us/ns * modify case * add case * add case * test: add test to ci. * modify case * fix: ci case issues (#31904) * enh: add operator reset func * fix: merge join reset issue * fix: memory issues * fix: add debug assert * fix: memory issues * fix: memory leak * fix: memory issues * fix taos log miss * fix: case issue * fix: case issue * fix: case issues * fix: drop dnode issue * fix: memory issues * fix: memory issues * fix: memory leak issues * fix: recalculate time range issue * fix: add debug log * fix: memory issues * fix: enable case asan * Update streamlist_for_ci.task * fix: case asan issue * fix: stream name issue * fix: external window compile issues * fix: deploy memory issue * fix: ahandle issue * fix: ahandle issue * fix: ahandle issue * fix: virtual table reader list issue * fix: log info * fix: msg error * fix: virtual table addr list issue * fix: memory issues * fix: memory leak issue * fix: memory issues * fix: memory free issues * fix: memory issues * fix: snode deploy issue * fix: mnode reader issue * fix: memory issues * fix: add debug test * enh: add ignore nodata trigger * fix: memory leaks * fix: configuration issue * fix: memory issue * fix: external window issue * fix: external window issues * fix: external window placeholder issue * fix: placeholder function init issues * fix: memory leak issue * fix: add debug log * fix: compile issues * fix: double free issue * fix: runner addr update issue * fix: msg rsp issue * fix: external window reset issue * fix: configuration issue * fix: deploy msg issue * fix: compile issue * fix: external window idx issue * fix: ci issues * fix: ci case issues * fix: drop dnode issue --------- Co-authored-by: huohong <sallyhuo@taosdata.com> * fix(stream): ci error * test: update test case. * feat: [TS-6100] Disable some failed UT. * feat: [TS-6100] Fix virtual table * test: add bug 5. * test: add test delete recalc to ci. * test: add bug 6. * test(stream): tobacco scene #TD-36514 * fix: reqCids,reqCols memory leak in SSTriggerRealtimeContext Co-authored-by: Tony Zhang <tonyzhang@taosdata.com> * test: add case stream6 * fix(stream): implement some pending features in trigger task * modify case * modify case * fix: case issues * modify case * test: add recalc for warter mark. * fix(stream): fix count window trigger of virtual tables * fix(stream): memory leak * test: fix run err. * test: add stream6 bug7 * fix: adjust format * test(stream): tobacco scene testing #TD-36514 * test: change bug7 with update window1 and 2 * test: add test bug 7. * case: restore write 3 window * fix: windows compile issue * fix: notify * test: add cases * modify case * test: update test case. * test(stream): toobacco scene testing #TD-36514 --------- Co-authored-by: Simon Guan <slguan@taosdata.com> Co-authored-by: plum-lihui <huili@taosdata.com> Co-authored-by: Alex Duan <417921451@qq.com> Co-authored-by: zelv01 <1101510017@qq.com> Co-authored-by: Jing Sima <simondominic9997@outlook.com> Co-authored-by: xiangyang guo <66111494+happyguoxy@users.noreply.github.com> Co-authored-by: wangmm0220 <wangmm0220@gmail.com> Co-authored-by: Haojun Liao <hjliao@taosdata.com> Co-authored-by: zyyang90 <zyyang@taosdata.com> Co-authored-by: Alex Duan <51781608+DuanKuanJun@users.noreply.github.com> Co-authored-by: facetosea <285808407@qq.com> Co-authored-by: Simon Guan <guanshengliang@qq.com> Co-authored-by: huohong <sallyhuo@taosdata.com> Co-authored-by: Li Hui <52318143+plum-lihui@users.noreply.github.com> Co-authored-by: Jinqing Kuang <kuangjinqingcn@gmail.com> Co-authored-by: xiao-77 <berylbao@taosdata.com> Co-authored-by: Zhixiao Bao <62235797+xiao-77@users.noreply.github.com> Co-authored-by: happyguoxy <happy_guoxy@163.com> Co-authored-by: Tony Zhang <34825804+Tony2h@users.noreply.github.com> Co-authored-by: Tony Zhang <tonyzhang@taosdata.com>
2025-07-16 06:42:16 +00:00
SWalReader* pRead = walOpenReader(pWal, 0);
2022-01-11 07:40:10 +00:00
ASSERT(pRead != NULL);
for (int i = 0; i < 1000; i++) {
2022-03-09 16:36:30 +00:00
int ver = taosRand() % 100;
2022-07-07 03:42:45 +00:00
code = walReadVer(pRead, ver);
2022-01-11 07:40:10 +00:00
ASSERT_EQ(code, 0);
// printf("rrbody: \n");
// for(int i = 0; i < pRead->pHead->head.len; i++) {
// printf("%d ", pRead->pHead->head.body[i]);
//}
// printf("\n");
ASSERT_EQ(pRead->pHead->head.version, ver);
ASSERT_EQ(pRead->curVersion, ver + 1);
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, ver);
int len = strlen(newStr);
2022-04-24 05:42:54 +00:00
ASSERT_EQ(pRead->pHead->head.bodyLen, len);
2022-01-11 07:40:10 +00:00
for (int j = 0; j < len; j++) {
EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
}
}
for (i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
2022-01-11 07:40:10 +00:00
ASSERT_EQ(code, 0);
}
2022-01-11 09:55:53 +00:00
for (int i = 0; i < 1000; i++) {
2022-03-09 16:36:30 +00:00
int ver = taosRand() % 200;
2022-07-07 03:42:45 +00:00
code = walReadVer(pRead, ver);
2022-01-11 09:55:53 +00:00
ASSERT_EQ(code, 0);
// printf("rrbody: \n");
// for(int i = 0; i < pRead->pHead->head.len; i++) {
// printf("%d ", pRead->pHead->head.body[i]);
//}
// printf("\n");
ASSERT_EQ(pRead->pHead->head.version, ver);
ASSERT_EQ(pRead->curVersion, ver + 1);
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, ver);
int len = strlen(newStr);
2022-04-24 05:42:54 +00:00
ASSERT_EQ(pRead->pHead->head.bodyLen, len);
2022-01-11 09:55:53 +00:00
for (int j = 0; j < len; j++) {
EXPECT_EQ(newStr[j], pRead->pHead->head.body[j]);
}
}
walCloseReader(pRead);
2022-01-11 07:40:10 +00:00
}
TEST_F(WalSkipLevel, restart) {
walResetEnv();
int code;
int i;
for (i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
TearDown();
SetUp();
}
TEST_F(WalSkipLevel, roll) {
int code;
int i;
for (i = 0; i < 100; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, 0);
code = walCommit(pWal, i);
}
walBeginSnapshot(pWal, i - 1, 0);
walEndSnapshot(pWal, false);
code = walAppendLog(pWal, 5, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_NE(code, 0);
for (; i < 200; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, 0);
code = walCommit(pWal, i);
}
code = walBeginSnapshot(pWal, i - 1, 0);
ASSERT_EQ(code, 0);
code = walEndSnapshot(pWal, false);
ASSERT_EQ(code, 0);
2024-12-24 09:34:57 +00:00
}
TEST_F(WalEncrypted, write) {
int code;
for (int i = 0; i < 100; i++) {
code = walAppendLog(pWal, i, i + 1, syncMeta, (void*)ranStr, ranStrLen, NULL);
2024-12-24 09:34:57 +00:00
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, i);
}
code = walSaveMeta(pWal);
ASSERT_EQ(code, 0);
}
TEST_F(WalKeepEnv, walSetKeepVersionBasic) {
walResetEnv();
int code;
// Test setting keep version
code = walSetKeepVersion(pWal, 50);
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->keepVersion, 50);
// Test setting different keep version
code = walSetKeepVersion(pWal, 100);
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->keepVersion, 100);
// Test invalid parameter (negative version)
code = walSetKeepVersion(pWal, -5);
ASSERT_NE(code, 0);
// Test NULL pointer
code = walSetKeepVersion(NULL, 50);
ASSERT_NE(code, 0);
}
TEST_F(WalCleanDeleteEnv, walSetKeepVersionWithDeletion) {
int code;
int i;
// Write 200 logs
for (i = 0; i < 200; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, 0);
ASSERT_EQ(pWal->vers.lastVer, i);
code = walCommit(pWal, i);
ASSERT_EQ(pWal->vers.commitVer, i);
}
// Set keep version to 50, so versions >= 50 should not be deleted
code = walSetKeepVersion(pWal, 50);
ASSERT_EQ(code, 0);
// Trigger snapshot, this should not delete logs with version >= 50
walBeginSnapshot(pWal, i - 1, 0);
ASSERT_EQ(pWal->vers.verInSnapshotting, i - 1);
walEndSnapshot(pWal, false);
ASSERT_EQ(pWal->vers.snapshotVer, i - 1);
ASSERT_EQ(pWal->vers.verInSnapshotting, -1);
// The firstVer should be no greater than 50
ASSERT_LE(pWal->vers.firstVer, 50);
// Continue writing more logs
for (; i < 300; i++) {
code = walAppendLog(pWal, i, 0, syncMeta, (void*)ranStr, ranStrLen, NULL);
ASSERT_EQ(code, 0);
code = walCommit(pWal, i);
ASSERT_EQ(pWal->vers.commitVer, i);
}
// Update keep version to 150
code = walSetKeepVersion(pWal, 150);
ASSERT_EQ(code, 0);
// Trigger another snapshot
code = walBeginSnapshot(pWal, i - 1, 0);
ASSERT_EQ(code, 0);
code = walEndSnapshot(pWal, false);
ASSERT_EQ(code, 0);
// The firstVer should be no greater than 150
ASSERT_LE(pWal->vers.firstVer, 150);
}
TEST_F(WalKeepEnv, walSetKeepVersionConcurrent) {
walResetEnv();
int code;
// Write some logs first
for (int i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Test concurrent calls to walSetKeepVersion with multiple threads
const int numThreads = 10;
const int callsPerThread = 100;
std::vector<std::thread> threads;
std::atomic<int> successCount(0);
for (int i = 0; i < numThreads; i++) {
threads.push_back(std::thread([this, i, &successCount, callsPerThread]() {
for (int j = 0; j < callsPerThread; j++) {
int64_t ver = i * callsPerThread + j;
int code = walSetKeepVersion(pWal, ver);
if (code == 0) {
successCount.fetch_add(1);
}
}
}));
}
// Wait for all threads to complete
for (auto& thread : threads) {
thread.join();
}
// All calls should succeed
ASSERT_EQ(successCount.load(), numThreads * callsPerThread);
// The final keepVersion should be one of the values set by the threads
// We can't predict which one exactly due to race conditions, but it should be valid
int64_t finalVersion = pWal->keepVersion;
ASSERT_GE(finalVersion, 0);
ASSERT_LT(finalVersion, numThreads * callsPerThread);
}
// Test walRenameCorruptedDir: delete first wal log file
TEST_F(WalRetentionEnv, corruptedDirDeleteFirstFile) {
walResetEnv();
int code;
// Enable walDeleteOnCorruption
bool oldVal = tsWalDeleteOnCorruption;
tsWalDeleteOnCorruption = true;
// Write logs to create multiple wal files
for (int i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Roll to create second file
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Roll to create third file
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 200; i < 300; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
TearDown();
// List and find the first log file in the WAL directory
TdDirPtr pDir = taosOpenDir(pathName);
ASSERT_NE(pDir, nullptr);
char firstLogFile[256] = {0};
TdDirEntryPtr pDirEntry;
while ((pDirEntry = taosReadDir(pDir)) != NULL) {
char* name = taosDirEntryBaseName(taosGetDirEntryName(pDirEntry));
if (strstr(name, ".log") != NULL) {
if (firstLogFile[0] == 0 || strcmp(name, firstLogFile) < 0) {
strncpy(firstLogFile, name, sizeof(firstLogFile) - 1);
}
}
}
taosCloseDir(&pDir);
ASSERT_NE(firstLogFile[0], 0);
char logFile[256];
snprintf(logFile, sizeof(logFile), "%s" TD_DIRSEP "%s", pathName, firstLogFile);
printf("Deleting first log file: %s\n", logFile);
taosRemoveFile(logFile);
// Wait for file system to complete the operation (especially important on Windows)
taosMsleep(100);
// Re-open should trigger walRenameCorruptedDir
SetUp();
// After rename, WAL should be recreated and empty
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.firstVer, 100);
ASSERT_EQ(pWal->vers.lastVer, 299);
// Verify old directory was renamed
char corruptedPath[300];
sprintf(corruptedPath, "%s.corrupted", pathName);
// The actual path will have a timestamp, so we just verify new dir exists and is empty
// Restore old value
tsWalDeleteOnCorruption = oldVal;
}
#ifndef WINDOWS
// Test walRenameCorruptedDir: delete middle wal log file
TEST_F(WalRetentionEnv, corruptedDirDeleteMiddleFile) {
walResetEnv();
int code;
bool oldVal = tsWalDeleteOnCorruption;
tsWalDeleteOnCorruption = true;
// Write logs to create multiple wal files
for (int i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 200; i < 300; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
TearDown();
// List all log files and find the middle one
TdDirPtr pDir = taosOpenDir(pathName);
ASSERT_NE(pDir, nullptr);
SArray* logFiles = taosArrayInit(10, sizeof(char[256]));
TdDirEntryPtr pDirEntry;
while ((pDirEntry = taosReadDir(pDir)) != NULL) {
char* name = taosDirEntryBaseName(taosGetDirEntryName(pDirEntry));
if (strstr(name, ".log") != NULL) {
char fileName[256];
strncpy(fileName, name, sizeof(fileName) - 1);
taosArrayPush(logFiles, fileName);
}
}
taosCloseDir(&pDir);
int fileCount = taosArrayGetSize(logFiles);
ASSERT_GT(fileCount, 1);
// Sort to ensure consistent ordering
taosArraySort(logFiles, (__compar_fn_t)strcmp);
// Delete the middle log file
char* middleFile = (char*)taosArrayGet(logFiles, fileCount / 2);
char logFile[256];
snprintf(logFile, sizeof(logFile), "%s" TD_DIRSEP "%s", pathName, middleFile);
printf("Deleting middle log file: %s\n", logFile);
taosRemoveFile(logFile);
taosArrayDestroy(logFiles);
// Wait for file system to complete the operation (especially important on Windows)
taosMsleep(100);
// Re-open should trigger walRenameCorruptedDir
SetUp();
// After rename, WAL should be recreated and empty
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.firstVer, -1);
ASSERT_EQ(pWal->vers.lastVer, -1);
tsWalDeleteOnCorruption = oldVal;
}
#endif
// Test walRenameCorruptedDir: delete last wal log file
TEST_F(WalRetentionEnv, corruptedDirDeleteLastFile) {
walResetEnv();
int code;
bool oldVal = tsWalDeleteOnCorruption;
tsWalDeleteOnCorruption = true;
// Write logs to create multiple wal files
for (int i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
for (int i = 200; i < 300; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
TearDown();
// List all log files and find the last one
TdDirPtr pDir = taosOpenDir(pathName);
ASSERT_NE(pDir, nullptr);
char lastLogFile[256] = {0};
TdDirEntryPtr pDirEntry;
while ((pDirEntry = taosReadDir(pDir)) != NULL) {
char* name = taosDirEntryBaseName(taosGetDirEntryName(pDirEntry));
if (strstr(name, ".log") != NULL) {
if (lastLogFile[0] == 0 || strcmp(name, lastLogFile) > 0) {
strncpy(lastLogFile, name, sizeof(lastLogFile) - 1);
}
}
}
taosCloseDir(&pDir);
ASSERT_NE(lastLogFile[0], 0);
char logFile[256];
snprintf(logFile, sizeof(logFile), "%s" TD_DIRSEP "%s", pathName, lastLogFile);
printf("Deleting last log file: %s\n", logFile);
taosRemoveFile(logFile);
// Wait for file system to complete the operation (especially important on Windows)
taosMsleep(100);
// Re-open should trigger walRenameCorruptedDir
SetUp();
// After rename, WAL should be recreated and empty
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.firstVer, 0);
ASSERT_EQ(pWal->vers.lastVer, 199);
tsWalDeleteOnCorruption = oldVal;
}
// Test for the bug: data loss after WAL corruption and recovery
// Scenario:
// 1. Write data to WAL
// 2. Simulate crash (close WAL)
// 3. Manually corrupt the last WAL file by appending garbage data
// 4. Restart and write new data
// 5. Restart again - verify new data is not lost
TEST_F(WalKeepEnv, corruptionRecoveryDataLoss) {
// Start with clean environment
walClose(pWal);
taosRemoveDir(pathName);
SWalCfg cfg = {0};
cfg.rollPeriod = -1;
cfg.segSize = -1;
cfg.retentionPeriod = 0;
cfg.retentionSize = 0;
cfg.level = TAOS_WAL_FSYNC;
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
int code;
// Step 1: Write initial data (version 0-9)
for (int i = 0; i < 10; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Get the log file path before closing
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
ASSERT_NE(pFileInfo, nullptr);
int64_t firstVer = pFileInfo->firstVer;
int64_t validFileSize = pFileInfo->fileSize;
char logFileName[256];
snprintf(logFileName, sizeof(logFileName), "%s" TD_DIRSEP "%020" PRId64 ".log", pathName, firstVer);
printf("Log file: %s, valid size: %" PRId64 "\n", logFileName, validFileSize);
// Step 2: Simulate crash - close WAL (but don't delete directory)
walClose(pWal);
pWal = NULL;
// Step 3: Manually corrupt the WAL file by appending garbage data
TdFilePtr pFile = taosOpenFile(logFileName, TD_FILE_WRITE | TD_FILE_APPEND);
if (pFile == nullptr) {
printf("Failed to open file: %s, error: %s\n", logFileName, strerror(errno));
}
ASSERT_NE(pFile, nullptr);
// Append 100 bytes of garbage data
char garbage[100];
memset(garbage, 0xFF, sizeof(garbage));
int64_t written = taosWriteFile(pFile, garbage, sizeof(garbage));
ASSERT_EQ(written, sizeof(garbage));
taosCloseFile(&pFile);
printf("Corrupted WAL file: %s (appended %d bytes of garbage)\n", logFileName, (int)sizeof(garbage));
// Step 4: Restart WAL and write new data (version 10-19)
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
// Verify that we recovered to version 9
ASSERT_EQ(pWal->vers.lastVer, 9);
// Write new data after corruption
for (int i = 10; i < 20; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
ASSERT_EQ(pWal->vers.lastVer, 19);
// Step 5: Restart again and verify all data (0-19) can be read
walClose(pWal);
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
// After restart, lastVer should still be 19
ASSERT_EQ(pWal->vers.lastVer, 19);
// Verify we can read all data from 0 to 19
SWalReader* pRead = walOpenReader(pWal, 0);
ASSERT_NE(pRead, nullptr);
for (int i = 0; i < 20; i++) {
code = walReadVer(pRead, i);
ASSERT_EQ(code, 0) << "Failed to read version " << i;
ASSERT_EQ(pRead->pHead->head.version, i);
char expectedStr[100];
sprintf(expectedStr, "%s-%d", ranStr, i);
int expectedLen = strlen(expectedStr);
ASSERT_EQ(pRead->pHead->head.bodyLen, expectedLen);
for (int j = 0; j < expectedLen; j++) {
EXPECT_EQ(expectedStr[j], pRead->pHead->head.body[j]);
}
}
walCloseReader(pRead);
printf("SUCCESS: All data (version 0-19) recovered correctly after corruption\n");
}
// Test for the bug with multiple files
// Verify that corruption in the last file doesn't affect data recovery
TEST_F(WalKeepEnv, corruptionRecoveryMultipleFiles) {
// Start with clean environment
walClose(pWal);
taosRemoveDir(pathName);
SWalCfg cfg = {0};
cfg.rollPeriod = -1;
cfg.segSize = -1;
cfg.retentionPeriod = 0;
cfg.retentionSize = 0;
cfg.level = TAOS_WAL_FSYNC;
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
int code;
// Write data to first file (version 0-99)
for (int i = 0; i < 100; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Roll to create second file
code = walRollImpl(pWal);
ASSERT_EQ(code, 0);
// Write data to second file (version 100-199)
for (int i = 100; i < 200; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
// Get the last log file path
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
ASSERT_NE(pFileInfo, nullptr);
int64_t lastFileFirstVer = pFileInfo->firstVer;
char logFileName[256];
snprintf(logFileName, sizeof(logFileName), "%s" TD_DIRSEP "%020" PRId64 ".log", pathName, lastFileFirstVer);
// Close WAL
walClose(pWal);
pWal = NULL;
// Corrupt the last file by appending garbage
TdFilePtr pFile = taosOpenFile(logFileName, TD_FILE_WRITE | TD_FILE_APPEND);
ASSERT_NE(pFile, nullptr);
char garbage[200];
memset(garbage, 0xAA, sizeof(garbage));
int64_t written = taosWriteFile(pFile, garbage, sizeof(garbage));
ASSERT_EQ(written, sizeof(garbage));
taosCloseFile(&pFile);
printf("Corrupted last WAL file: %s\n", logFileName);
// Restart and write new data
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.lastVer, 199);
// Write new data (version 200-249)
for (int i = 200; i < 250; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
ASSERT_EQ(pWal->vers.lastVer, 249);
// Restart and verify all data
walClose(pWal);
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.lastVer, 249);
// Verify we can read all data
SWalReader* pRead = walOpenReader(pWal, 0);
ASSERT_NE(pRead, nullptr);
for (int i = 0; i < 250; i++) {
code = walReadVer(pRead, i);
ASSERT_EQ(code, 0) << "Failed to read version " << i;
ASSERT_EQ(pRead->pHead->head.version, i);
}
walCloseReader(pRead);
printf("SUCCESS: All data recovered correctly with multiple files\n");
}
// Test behavior when corruption is in the middle of a WAL file (file size unchanged).
// The repair logic triggers a full scan when fileSize != meta fileSize (e.g. appended garbage).
// For in-place overwrites the file size matches, so the scan is skipped and lastVer stays at 19.
// Reading entries at/after the corrupted offset will fail with a checksum error.
TEST_F(WalKeepEnv, corruptionInMiddleOfFile) {
// Start with clean environment
walClose(pWal);
taosRemoveDir(pathName);
SWalCfg cfg = {0};
cfg.rollPeriod = -1;
cfg.segSize = -1;
cfg.retentionPeriod = 0;
cfg.retentionSize = 0;
cfg.level = TAOS_WAL_FSYNC;
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
int code;
// Write initial data (version 0-19)
for (int i = 0; i < 20; i++) {
char newStr[100];
sprintf(newStr, "%s-%d", ranStr, i);
int len = strlen(newStr);
code = walAppendLog(pWal, i, 0, syncMeta, newStr, len, NULL);
ASSERT_EQ(code, 0);
}
SWalFileInfo* pFileInfo = walGetCurFileInfo(pWal);
ASSERT_NE(pFileInfo, nullptr);
int64_t validFileSize = pFileInfo->fileSize;
int64_t firstVer = pFileInfo->firstVer;
char logFileName[256];
snprintf(logFileName, sizeof(logFileName), "%s" TD_DIRSEP "%020" PRId64 ".log", pathName, firstVer);
// Close WAL
walClose(pWal);
pWal = NULL;
// Corrupt by overwriting middle of file with garbage (file size unchanged)
TdFilePtr pFile = taosOpenFile(logFileName, TD_FILE_WRITE);
ASSERT_NE(pFile, nullptr);
int64_t corruptOffset = validFileSize / 2;
taosLSeekFile(pFile, corruptOffset, SEEK_SET);
char garbage[50];
memset(garbage, 0xBB, sizeof(garbage));
taosWriteFile(pFile, garbage, sizeof(garbage));
taosCloseFile(&pFile);
printf("Corrupted middle of WAL file at offset %" PRId64 "\n", corruptOffset);
// Restart - file size matches meta, so repair scan is skipped; lastVer stays at 19
pWal = walOpen(pathName, &cfg);
ASSERT_NE(pWal, nullptr);
ASSERT_EQ(pWal->vers.lastVer, 19);
// Reading entries at/after the corrupted offset will fail
SWalReader* pRead = walOpenReader(pWal, 0);
ASSERT_NE(pRead, nullptr);
// Version 0 (before corruption) should be readable
code = walReadVer(pRead, 0);
ASSERT_EQ(code, 0);
// At least one entry in the corrupted region should fail
bool foundError = false;
for (int i = 0; i < 20; i++) {
if (walReadVer(pRead, i) != 0) {
foundError = true;
printf("Read error at version %d as expected\n", i);
break;
}
}
ASSERT_TRUE(foundError) << "Expected at least one read failure due to middle corruption";
walCloseReader(pRead);
printf("SUCCESS: Middle corruption detected at read time as expected\n");
}