Merge branch '3.0' into feat/TD-30053-3.0

This commit is contained in:
kailixu 2024-06-07 10:28:22 +08:00
commit 7de2928c86
47 changed files with 313 additions and 103 deletions

View File

@ -1,5 +1,5 @@
if (! "RJDBC" %in% installed.packages()[, "Package"]) {
install.packages('RJDBC', repos='http://cran.us.r-project.org')
install.packages('RJDBC', repos='http://mirrors.tuna.tsinghua.edu.cn/CRAN')
}
# ANCHOR: demo

View File

@ -224,13 +224,14 @@
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_ENCRYPT_KEY, "create-encrypt-key", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_S3MIGRATE_DB, "s3migrate-db", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_S3MIGRATE_DB_TIMER, "s3migrate-db-tmr", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_UNUSED2, "mnd-unused2", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TSMA, "create-tsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TSMA, "drop-tsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_STB_DROP, "drop-stb", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_GET_TABLE_TSMA, "get-table-tsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_GET_TSMA, "get-tsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TB_WITH_TSMA, "drop-tb-with-tsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_MND_MSG)
TD_NEW_MSG_SEG(TDMT_VND_MSG) // 2<<8
@ -275,7 +276,6 @@
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_HASHRANGE, "alter-hashrange", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_COMPACT, "vnode-compact", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DROP_TTL_TABLE, "vnode-drop-ttl-stb", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_FETCH_TTL_EXPIRED_TBS, "vnode-fetch-ttl-expired-tbs", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_TRIM, "vnode-trim", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_COMMIT, "vnode-commit", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_CREATE_INDEX, "vnode-create-index", NULL, NULL)
@ -286,6 +286,7 @@
TD_DEF_MSG_TYPE(TDMT_VND_S3MIGRATE, "vnode-s3migrate", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ARB_HEARTBEAT, "vnode-arb-hb", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ARB_CHECK_SYNC, "vnode-arb-check-sync", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_FETCH_TTL_EXPIRED_TBS, "vnode-fetch-ttl-expired-tbs", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_MAX_MSG, "vnd-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_VND_MSG)
@ -318,11 +319,12 @@
TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_PAUSE, "stream-task-pause", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RESUME, "stream-task-resume", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_STOP, "stream-task-stop", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_MAX_MSG, "stream-max", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_UNUSED, "stream-unused", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_CREATE, "stream-create", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_DROP, "stream-drop", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE_TRIGGER, "stream-retri-trigger", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_STREAM_MAX_MSG, "stream-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_STREAM_MSG)
TD_NEW_MSG_SEG(TDMT_MON_MSG) //5 << 8
@ -369,8 +371,9 @@
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TASK_UPDATE, "vnode-stream-update", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TASK_RESET, "vnode-stream-reset", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TASK_CHECK, "vnode-stream-task-check", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_MAX_MSG, "vnd-stream-max", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_UNUSED, "vnd-stream-unused", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_GET_STREAM_PROGRESS, "vnd-stream-progress", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_STREAM_MAX_MSG, "vnd-stream-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_VND_STREAM_MSG)
TD_NEW_MSG_SEG(TDMT_VND_TMQ_MSG) //8 << 8

View File

@ -45,6 +45,7 @@ extern "C" {
#define WAL_SCAN_BUF_SIZE (1024 * 1024 * 3)
typedef enum {
TAOS_WAL_SKIP = 0,
TAOS_WAL_WRITE = 1,
TAOS_WAL_FSYNC = 2,
} EWalType;
@ -145,17 +146,18 @@ typedef struct SWalReader SWalReader;
// todo hide this struct
struct SWalReader {
SWal *pWal;
int64_t readerId;
TdFilePtr pLogFile;
TdFilePtr pIdxFile;
int64_t curFileFirstVer;
int64_t curVersion;
int64_t skipToVersion; // skip data and jump to destination version, usually used by stream resume ignoring untreated data
SWal *pWal;
int64_t readerId;
TdFilePtr pLogFile;
TdFilePtr pIdxFile;
int64_t curFileFirstVer;
int64_t curVersion;
int64_t skipToVersion; // skip data and jump to destination version, usually used by stream resume ignoring untreated
// data
int64_t capacity;
TdThreadMutex mutex;
SWalFilterCond cond;
SWalCkHead *pHead;
SWalCkHead *pHead;
};
// module initialization
@ -198,7 +200,7 @@ SWalReader *walOpenReader(SWal *, SWalFilterCond *pCond, int64_t id);
void walCloseReader(SWalReader *pRead);
void walReadReset(SWalReader *pReader);
int32_t walReadVer(SWalReader *pRead, int64_t ver);
void decryptBody(SWalCfg* cfg, SWalCkHead* pHead, int32_t plainBodyLen, const char* func);
void decryptBody(SWalCfg *cfg, SWalCkHead *pHead, int32_t plainBodyLen, const char *func);
int32_t walReaderSeekVer(SWalReader *pRead, int64_t ver);
int32_t walNextValidMsg(SWalReader *pRead);
int64_t walReaderGetCurrentVer(const SWalReader *pReader);
@ -206,7 +208,7 @@ int64_t walReaderGetValidFirstVer(const SWalReader *pReader);
int64_t walReaderGetSkipToVersion(SWalReader *pReader);
void walReaderSetSkipToVersion(SWalReader *pReader, int64_t ver);
void walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *ever);
void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal* pOffset);
void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal *pOffset);
// only for tq usage
int32_t walFetchHead(SWalReader *pRead, int64_t ver);

View File

@ -373,7 +373,7 @@ typedef enum ELogicConditionType {
#define TSDB_MIN_FSYNC_PERIOD 0
#define TSDB_MAX_FSYNC_PERIOD 180000 // millisecond
#define TSDB_DEFAULT_FSYNC_PERIOD 3000 // three second
#define TSDB_MIN_WAL_LEVEL 1
#define TSDB_MIN_WAL_LEVEL 0
#define TSDB_MAX_WAL_LEVEL 2
#define TSDB_DEFAULT_WAL_LEVEL 1
#define TSDB_MIN_PRECISION TSDB_TIME_PRECISION_MILLI

View File

@ -826,7 +826,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
offRows->vgId = pVg->vgId;
offRows->rows = pVg->numOfRows;
offRows->offset = pVg->offsetInfo.endOffset;
offRows->ever = pVg->offsetInfo.walVerEnd;
offRows->ever = pVg->offsetInfo.walVerEnd == -1 ? 0 : pVg->offsetInfo.walVerEnd;
char buf[TSDB_OFFSET_LEN] = {0};
tFormatOffset(buf, TSDB_OFFSET_LEN, &offRows->offset);
tscDebug("consumer:0x%" PRIx64 ",report offset, group:%s vgId:%d, offset:%s/%" PRId64 ", rows:%" PRId64,

View File

@ -796,7 +796,13 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
static bool mndCheckDbConflict(const char *conflict, STrans *pTrans) {
if (conflict[0] == 0) return false;
if (strcasecmp(conflict, pTrans->dbname) == 0 || strcasecmp(conflict, pTrans->stbname) == 0) return true;
if (strcasecmp(conflict, pTrans->dbname) == 0) return true;
return false;
}
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
if (conflict[0] == 0) return false;
if (strcasecmp(conflict, pTrans->stbname) == 0) return true;
return false;
}
@ -816,17 +822,17 @@ static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
if (mndCheckDbConflict(pNew->dbname, pTrans)) conflict = true;
if (mndCheckDbConflict(pNew->stbname, pTrans)) conflict = true;
if (mndCheckStbConflict(pNew->stbname, pTrans)) conflict = true;
}
}
if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
if (pTrans->conflict == TRN_CONFLICT_DB) {
if (mndCheckDbConflict(pNew->dbname, pTrans)) conflict = true;
if (mndCheckDbConflict(pNew->stbname, pTrans)) conflict = true;
if (mndCheckStbConflict(pNew->stbname, pTrans)) conflict = true;
}
if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
if (mndCheckDbConflict(pNew->stbname, pTrans)) conflict = true; // for stb
if (mndCheckStbConflict(pNew->stbname, pTrans)) conflict = true; // for stb
}
}

View File

@ -6771,6 +6771,15 @@ static int32_t checkDbEnumOption(STranslateContext* pCxt, const char* pName, int
return TSDB_CODE_SUCCESS;
}
static int32_t checkDbEnumOption3(STranslateContext* pCxt, const char* pName, int32_t val, int32_t v1, int32_t v2,
int32_t v3) {
if (val >= 0 && val != v1 && val != v2 && val != v3) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option %s: %d, only %d, %d, %d allowed", pName, val, v1, v2, v3);
}
return TSDB_CODE_SUCCESS;
}
static int32_t checkDbRetentionsOption(STranslateContext* pCxt, SNodeList* pRetentions, int8_t precision) {
if (NULL == pRetentions) {
return TSDB_CODE_SUCCESS;
@ -6914,7 +6923,12 @@ static int32_t checkOptionsDependency(STranslateContext* pCxt, const char* pDbNa
if ((pOptions->replica == 2) ^ (pOptions->withArbitrator == TSDB_MAX_DB_WITH_ARBITRATOR)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid database option, with_arbitrator should be used with replica 2");
"Invalid option, with_arbitrator should be used with replica 2");
}
if (pOptions->replica > 1 && pOptions->walLevel == 0) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option, wal_level 0 should be used with replica 1");
}
return TSDB_CODE_SUCCESS;
@ -6978,7 +6992,8 @@ static int32_t checkDatabaseOptions(STranslateContext* pCxt, const char* pDbName
code = checkDbStrictOption(pCxt, pOptions);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbEnumOption(pCxt, "walLevel", pOptions->walLevel, TSDB_MIN_WAL_LEVEL, TSDB_MAX_WAL_LEVEL);
code = checkDbEnumOption3(pCxt, "walLevel", pOptions->walLevel, TSDB_MIN_WAL_LEVEL, TSDB_DEFAULT_WAL_LEVEL,
TSDB_MAX_WAL_LEVEL);
}
if (TSDB_CODE_SUCCESS == code) {
code = checkDbRangeOption(pCxt, "vgroups", pOptions->numOfVgroups, TSDB_MIN_VNODES_PER_DB, TSDB_MAX_VNODES_PER_DB);
@ -7252,6 +7267,15 @@ static void buildAlterDbReq(STranslateContext* pCxt, SAlterDatabaseStmt* pStmt,
}
static int32_t translateAlterDatabase(STranslateContext* pCxt, SAlterDatabaseStmt* pStmt) {
if (pStmt->pOptions->walLevel == 0) {
SDbCfgInfo dbCfg = {0};
int32_t code = getDBCfg(pCxt, pStmt->dbName, &dbCfg);
if (TSDB_CODE_SUCCESS == code && dbCfg.replications > 1) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DB_OPTION,
"Invalid option, wal_level 0 should be used with replica 1");
}
}
int32_t code = checkDatabaseOptions(pCxt, pStmt->dbName, pStmt->pOptions);
if (TSDB_CODE_SUCCESS != code) {
return code;

View File

@ -335,7 +335,7 @@ TEST_F(ParserInitialATest, alterDatabaseSemanticCheck) {
run("ALTER DATABASE test KEEP 1000000000s", TSDB_CODE_PAR_INVALID_DB_OPTION);
run("ALTER DATABASE test KEEP 1w", TSDB_CODE_PAR_INVALID_DB_OPTION);
run("ALTER DATABASE test PAGES 63", TSDB_CODE_PAR_INVALID_DB_OPTION);
run("ALTER DATABASE test WAL_LEVEL 0", TSDB_CODE_PAR_INVALID_DB_OPTION);
//run("ALTER DATABASE test WAL_LEVEL 0", TSDB_CODE_PAR_INVALID_DB_OPTION);
run("ALTER DATABASE test WAL_LEVEL 3", TSDB_CODE_PAR_INVALID_DB_OPTION);
//run("ALTER DATABASE test REPLICA 2", TSDB_CODE_PAR_INVALID_DB_OPTION);
run("ALTER DATABASE test STT_TRIGGER 0", TSDB_CODE_PAR_INVALID_DB_OPTION);

View File

@ -20,7 +20,6 @@
#include "tutil.h"
#include "walInt.h"
bool FORCE_INLINE walLogExist(SWal* pWal, int64_t ver) {
return !walIsEmpty(pWal) && walGetFirstVer(pWal) <= ver && walGetLastVer(pWal) >= ver;
}
@ -154,7 +153,7 @@ static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx) {
// validate body
int32_t cryptedBodyLen = logContent->head.bodyLen;
if(pWal->cfg.encryptAlgorithm == DND_CA_SM4){
if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
cryptedBodyLen = ENCRYPTED_LEN(cryptedBodyLen);
}
recordLen = walCkHeadSz + cryptedBodyLen;
@ -226,7 +225,7 @@ static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx) {
goto _err;
}
if (taosFsyncFile(pFile) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosFsyncFile(pFile) < 0) {
wError("failed to fsync file due to %s. file:%s", strerror(errno), fnameStr);
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
@ -626,7 +625,7 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
int32_t plainBodyLen = ckHead.head.bodyLen;
int32_t cryptedBodyLen = plainBodyLen;
if(pWal->cfg.encryptAlgorithm == DND_CA_SM4){
if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
cryptedBodyLen = ENCRYPTED_LEN(cryptedBodyLen);
}
idxEntry.offset += sizeof(SWalCkHead) + cryptedBodyLen;
@ -636,7 +635,7 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
pWal->cfg.vgId, terrstr(), idxEntry.ver, idxEntry.offset, fLogNameStr);
goto _err;
}
if (taosWriteFile(pIdxFile, &idxEntry, sizeof(SWalIdxEntry)) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosWriteFile(pIdxFile, &idxEntry, sizeof(SWalIdxEntry)) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, failed to append file since %s. file:%s", pWal->cfg.vgId, terrstr(), fnameStr);
goto _err;
@ -644,7 +643,7 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
count++;
}
if (taosFsyncFile(pIdxFile) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosFsyncFile(pIdxFile) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, faild to fsync file since %s. file:%s", pWal->cfg.vgId, terrstr(), fnameStr);
goto _err;
@ -880,13 +879,13 @@ int walSaveMeta(SWal* pWal) {
int n;
// fsync the idx and log file at first to ensure validity of meta
if (taosFsyncFile(pWal->pIdxFile) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosFsyncFile(pWal->pIdxFile) < 0) {
wError("vgId:%d, failed to sync idx file due to %s", pWal->cfg.vgId, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (taosFsyncFile(pWal->pLogFile) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosFsyncFile(pWal->pLogFile) < 0) {
wError("vgId:%d, failed to sync log file due to %s", pWal->cfg.vgId, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
@ -901,7 +900,8 @@ int walSaveMeta(SWal* pWal) {
return -1;
}
TdFilePtr pMetaFile = taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
TdFilePtr pMetaFile =
taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
if (pMetaFile == NULL) {
wError("vgId:%d, failed to open file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr);
terrno = TAOS_SYSTEM_ERROR(errno);
@ -910,13 +910,13 @@ int walSaveMeta(SWal* pWal) {
char* serialized = walMetaSerialize(pWal);
int len = strlen(serialized);
if (len != taosWriteFile(pMetaFile, serialized, len)) {
if (pWal->cfg.level != TAOS_WAL_SKIP && len != taosWriteFile(pMetaFile, serialized, len)) {
wError("vgId:%d, failed to write file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr);
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile(pMetaFile) < 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosFsyncFile(pMetaFile) < 0) {
wError("vgId:%d, failed to sync file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr);
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;

View File

@ -79,8 +79,7 @@ int64_t walChangeWrite(SWal* pWal, int64_t ver) {
TdFilePtr pIdxTFile, pLogTFile;
char fnameStr[WAL_FILE_LEN];
if (pWal->pLogFile != NULL) {
code = taosFsyncFile(pWal->pLogFile);
if (code != 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
@ -91,8 +90,7 @@ int64_t walChangeWrite(SWal* pWal, int64_t ver) {
}
}
if (pWal->pIdxFile != NULL) {
code = taosFsyncFile(pWal->pIdxFile);
if (code != 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}

View File

@ -13,12 +13,12 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "crypt.h"
#include "os.h"
#include "taoserror.h"
#include "tchecksum.h"
#include "tglobal.h"
#include "walInt.h"
#include "crypt.h"
int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
taosThreadMutexLock(&pWal->mutex);
@ -295,9 +295,9 @@ int32_t walEndSnapshot(SWal *pWal) {
ver = TMAX(ver - pWal->vers.logRetention, pWal->vers.firstVer - 1);
// compatible mode for refVer
bool hasTopic = false;
bool hasTopic = false;
int64_t refVer = INT64_MAX;
void *pIter = NULL;
void *pIter = NULL;
while (1) {
pIter = taosHashIterate(pWal->pRefHash, pIter);
if (pIter == NULL) break;
@ -396,8 +396,7 @@ int32_t walRollImpl(SWal *pWal) {
int32_t code = 0;
if (pWal->pIdxFile != NULL) {
code = taosFsyncFile(pWal->pIdxFile);
if (code != 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pIdxFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
@ -409,8 +408,7 @@ int32_t walRollImpl(SWal *pWal) {
}
if (pWal->pLogFile != NULL) {
code = taosFsyncFile(pWal->pLogFile);
if (code != 0) {
if (pWal->cfg.level != TAOS_WAL_SKIP && (code = taosFsyncFile(pWal->pLogFile)) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto END;
}
@ -510,12 +508,15 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
wDebug("vgId:%d, wal write log %" PRId64 ", msgType: %s, cksum head %u cksum body %u", pWal->cfg.vgId, index,
TMSG_INFO(msgType), pWal->writeHead.cksumHead, pWal->writeHead.cksumBody);
code = walWriteIndex(pWal, index, offset);
if (code < 0) {
goto END;
if (pWal->cfg.level != TAOS_WAL_SKIP) {
code = walWriteIndex(pWal, index, offset);
if (code < 0) {
goto END;
}
}
if (taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead)) != sizeof(SWalCkHead)) {
if (pWal->cfg.level != TAOS_WAL_SKIP &&
taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead)) != sizeof(SWalCkHead)) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
@ -524,17 +525,17 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
}
int32_t cyptedBodyLen = plainBodyLen;
char* buf = (char*)body;
char* newBody = NULL;
char* newBodyEncrypted = NULL;
char *buf = (char *)body;
char *newBody = NULL;
char *newBodyEncrypted = NULL;
if(pWal->cfg.encryptAlgorithm == DND_CA_SM4){
if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
cyptedBodyLen = ENCRYPTED_LEN(cyptedBodyLen);
newBody = taosMemoryMalloc(cyptedBodyLen);
if(newBody == NULL){
if (newBody == NULL) {
wError("vgId:%d, file:%" PRId64 ".log, failed to malloc since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
strerror(errno));
code = -1;
goto END;
}
@ -542,11 +543,11 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
memcpy(newBody, body, plainBodyLen);
newBodyEncrypted = taosMemoryMalloc(cyptedBodyLen);
if(newBodyEncrypted == NULL){
if (newBodyEncrypted == NULL) {
wError("vgId:%d, file:%" PRId64 ".log, failed to malloc since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
strerror(errno));
code = -1;
if(newBody != NULL) taosMemoryFreeClear(newBody);
if (newBody != NULL) taosMemoryFreeClear(newBody);
goto END;
}
@ -559,29 +560,29 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
int32_t count = CBC_Encrypt(&opts);
//wDebug("vgId:%d, file:%" PRId64 ".log, index:%" PRId64 ", CBC_Encrypt cryptedBodyLen:%d, plainBodyLen:%d, %s",
// pWal->cfg.vgId, walGetLastFileFirstVer(pWal), index, count, plainBodyLen, __FUNCTION__);
// wDebug("vgId:%d, file:%" PRId64 ".log, index:%" PRId64 ", CBC_Encrypt cryptedBodyLen:%d, plainBodyLen:%d, %s",
// pWal->cfg.vgId, walGetLastFileFirstVer(pWal), index, count, plainBodyLen, __FUNCTION__);
buf = newBodyEncrypted;
}
if (taosWriteFile(pWal->pLogFile, (char *)buf, cyptedBodyLen) != cyptedBodyLen) {
if (pWal->cfg.level != TAOS_WAL_SKIP && taosWriteFile(pWal->pLogFile, (char *)buf, cyptedBodyLen) != cyptedBodyLen) {
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
code = -1;
if(pWal->cfg.encryptAlgorithm == DND_CA_SM4){
if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
taosMemoryFreeClear(newBody);
taosMemoryFreeClear(newBodyEncrypted);
}
goto END;
}
if(pWal->cfg.encryptAlgorithm == DND_CA_SM4){
if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
taosMemoryFreeClear(newBody);
taosMemoryFreeClear(newBodyEncrypted);
//wInfo("vgId:%d, free newBody newBodyEncrypted %s",
// pWal->cfg.vgId, __FUNCTION__);
taosMemoryFreeClear(newBodyEncrypted);
// wInfo("vgId:%d, free newBody newBodyEncrypted %s",
// pWal->cfg.vgId, __FUNCTION__);
}
// set status
@ -693,6 +694,10 @@ int32_t walWrite(SWal *pWal, int64_t index, tmsg_t msgType, const void *body, in
}
void walFsync(SWal *pWal, bool forceFsync) {
if (pWal->cfg.level == TAOS_WAL_SKIP) {
return;
}
taosThreadMutexLock(&pWal->mutex);
if (forceFsync || (pWal->cfg.level == TAOS_WAL_FSYNC && pWal->cfg.fsyncPeriod == 0)) {
wTrace("vgId:%d, fileId:%" PRId64 ".log, do fsync", pWal->cfg.vgId, walGetCurFileFirstVer(pWal));

View File

@ -0,0 +1,110 @@
import taos
import sys
import os
import subprocess
import glob
import shutil
import time
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.srvCtl import *
from frame.caseBase import *
from frame import *
from frame.autogen import *
# from frame.server.dnodes import *
# from frame.server.cluster import *
class TDTestCase(TBase):
updatecfgDict = {
'slowLogScope':"query"
}
def init(self, conn, logSql, replicaVar=3):
super(TDTestCase, self).init(conn, logSql, replicaVar=3, db="snapshot", checkColName="c1")
self.valgrind = 0
self.childtable_count = 10
# tdSql.init(conn.cursor())
tdSql.init(conn.cursor(), logSql) # output sql.txt file
def run(self):
tdSql.prepare()
autoGen = AutoGen()
autoGen.create_db(self.db, 2, 3)
tdSql.execute(f"use {self.db}")
autoGen.create_stable(self.stb, 5, 10, 8, 8)
autoGen.create_child(self.stb, "d", self.childtable_count)
autoGen.insert_data(1000)
tdSql.execute(f"flush database {self.db}")
sc.dnodeStop(3)
# clusterDnodes.stoptaosd(1)
# clusterDnodes.starttaosd(3)
# time.sleep(5)
# clusterDnodes.stoptaosd(2)
# clusterDnodes.starttaosd(1)
# time.sleep(5)
autoGen.insert_data(5000, True)
self.flushDb(True)
# wait flush operation over
time.sleep(5)
# sql = 'show vnodes;'
# while True:
# bFinish = True
# param_list = tdSql.query(sql, row_tag=True)
# for param in param_list:
# if param[3] == 'leading' or param[3] == 'following':
# bFinish = False
# break
# if bFinish:
# break
self.snapshotAgg()
time.sleep(10)
sc.dnodeStopAll()
for i in range(1, 4):
path = clusterDnodes.getDnodeDir(i)
dnodesRootDir = os.path.join(path,"data","vnode", "vnode*")
dirs = glob.glob(dnodesRootDir)
for dir in dirs:
if os.path.isdir(dir):
self.remove_directory(os.path.join(dir, "wal"))
sc.dnodeStart(1)
sc.dnodeStart(2)
sc.dnodeStart(3)
sql = "show vnodes;"
time.sleep(10)
while True:
bFinish = True
param_list = tdSql.query(sql, row_tag=True)
for param in param_list:
if param[3] == 'offline':
tdLog.exit(
"dnode synchronous fail dnode id: %d, vgroup id:%d status offline" % (param[0], param[1]))
if param[3] == 'leading' or param[3] == 'following':
bFinish = False
break
if bFinish:
break
self.timestamp_step = 1000
self.insert_rows = 6000
self.checkInsertCorrect()
self.checkAggCorrect()
def remove_directory(self, directory):
try:
shutil.rmtree(directory)
tdLog.debug("delete dir: %s " % (directory))
except OSError as e:
tdLog.exit("delete fail dir: %s " % (directory))
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

View File

@ -19,5 +19,3 @@ import sys
import os
import time
import datetime

28
tests/army/frame/eutil.py Normal file
View File

@ -0,0 +1,28 @@
###################################################################
# Copyright (c) 2023 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
#
# about tools funciton extension
#
import sys
import os
import time
import datetime
import psutil
# cpu frequent as random
def cpuRand(max):
decimal = int(str(psutil.cpu_freq().current).split(".")[1])
return decimal % max

View File

@ -19,6 +19,7 @@ import taos
import frame
import frame.etool
import frame.eos
import frame.eutil
from frame.log import *
from frame.cases import *
@ -46,17 +47,20 @@ for test:
class TDTestCase(TBase):
index = eutil.cpuRand(20) + 1
bucketName = f"ci-bucket{index}"
updatecfgDict = {
"supportVnodes":"1000",
's3EndPoint': 'http://192.168.1.52:9000',
's3AccessKey': 'zOgllR6bSnw2Ah3mCNel:cdO7oXAu3Cqdb1rUdevFgJMi0LtRwCXdWKQx4bhX',
's3BucketName': 'ci-bucket',
's3BucketName': f'{bucketName}',
's3PageCacheSize': '10240',
"s3UploadDelaySec": "10",
's3MigrateIntervalSec': '600',
's3MigrateEnabled': '1'
}
tdLog.info(f"assign bucketName is {bucketName}\n")
maxFileSize = (128 + 10) * 1014 * 1024 # add 10M buffer
def insertData(self):
@ -241,10 +245,9 @@ class TDTestCase(TBase):
#
def preDb(self, vgroups):
cnt = int(time.time())%3 + 1
cnt = int(time.time())%2 + 1
for i in range(cnt):
vg = int(time.time()*1000)%10 + 1
vg = eutil.cpuRand(9) + 1
sql = f"create database predb vgroups {vg}"
tdSql.execute(sql, show=True)
sql = "drop database predb"

View File

@ -10,26 +10,26 @@
#
# army-test
#
,,y,army,./pytest.sh python3 ./test.py -f enterprise/multi-level/mlevel_basic.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f enterprise/db-encrypt/basic.py
,,n,army,python3 ./test.py -f enterprise/s3/s3Basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/cluster/snapshot.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f community/query/function/test_func_elapsed.py
,,y,army,./pytest.sh python3 ./test.py -f community/query/test_join.py
,,y,army,./pytest.sh python3 ./test.py -f community/query/test_compare.py
,,y,army,./pytest.sh python3 ./test.py -f community/insert/test_column_tag_boundary.py
,,y,army,./pytest.sh python3 ./test.py -f community/query/fill/fill_desc.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f community/cluster/incSnapshot.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/query/query_basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/query/accuracy/test_query_accuracy.py
,,y,army,./pytest.sh python3 ./test.py -f community/insert/insert_basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/cluster/splitVgroupByLearner.py -N 3
,,n,army,python3 ./test.py -f community/cmdline/fullopt.py
,,n,army,python3 ./test.py -f community/query/show.py -N 3
,,n,army,python3 ./test.py -f enterprise/alter/alterConfig.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/query/subquery/subqueryBugs.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f community/storage/oneStageComp.py -N 3 -L 3 -D 1
,,y,army,./pytest.sh python3 ./test.py -f community/storage/compressBasic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f multi-level/mlevel_basic.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f db-encrypt/basic.py
,,n,army,python3 ./test.py -f s3/s3Basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f cluster/snapshot.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f query/function/test_func_elapsed.py
,,y,army,./pytest.sh python3 ./test.py -f query/test_join.py
,,y,army,./pytest.sh python3 ./test.py -f query/test_compare.py
,,y,army,./pytest.sh python3 ./test.py -f insert/test_column_tag_boundary.py
,,y,army,./pytest.sh python3 ./test.py -f query/fill/fill_desc.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f cluster/incSnapshot.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f query/query_basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f query/accuracy/test_query_accuracy.py
,,y,army,./pytest.sh python3 ./test.py -f insert/insert_basic.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f cluster/splitVgroupByLearner.py -N 3
,,n,army,python3 ./test.py -f cmdline/fullopt.py
,,n,army,python3 ./test.py -f query/show.py -N 3
,,n,army,python3 ./test.py -f alter/alterConfig.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f query/subquery/subqueryBugs.py -N 3
,,y,army,./pytest.sh python3 ./test.py -f storage/oneStageComp.py -N 3 -L 3 -D 1
,,y,army,./pytest.sh python3 ./test.py -f storage/compressBasic.py -N 3
#
# system test

View File

@ -49,7 +49,7 @@ print ============= create database with all options
# | KEEP value [max(1d ~ 365000d), default: 1d, unit may be minut/hour/day]
# | PRECISION ['ms' | 'us' | 'ns', default: ms]
# | REPLICA value [1 | 3, default: 1]
# | WAL_LEVEL value [1 | 2, default: 1]
# | WAL_LEVEL value [0 | 1 | 2, default: 1]
# | VGROUPS value [default: 2]
# | SINGLE_STABLE [0 | 1, default: ]
#
@ -404,7 +404,7 @@ endi
sql drop database db
sql_error create database db WAL_LEVEL 3
sql_error create database db WAL_LEVEL -1
sql_error create database db WAL_LEVEL 0
#sql_error create database db WAL_LEVEL 0
print ====> VGROUPS value [1~4096, default: 2]
sql create database db VGROUPS 1

View File

@ -225,7 +225,7 @@ sql_error create database $db ctime 29
sql_error create database $db ctime 40961
# wal {0, 2}
sql_error create database testwal wal_level 0
#sql_error create database testwal wal_level 0
sql select * from information_schema.ins_databases
if $rows != 2 then
return -1

View File

@ -148,7 +148,7 @@ class TDTestCase:
@property
def fsync_create_err(self):
return [
"create database db1 wal_level 0",
#"create database db1 wal_level 0",
"create database db1 wal_level 3",
"create database db1 wal_level null",
"create database db1 wal_level true",
@ -162,7 +162,7 @@ class TDTestCase:
@property
def fsync_alter_err(self):
return [
"alter database db1 wal_level 0",
#"alter database db1 wal_level 0",
"alter database db1 wal_level 3",
"alter database db1 wal_level null",
"alter database db1 wal_level true",

View File

@ -257,10 +257,10 @@ class TDTestCase:
os.system(f'taos -f {sql_file}')
tdSql.query('select count(c_1) from d2.t2 where c_1 < 10', queryTimes=1)
tdSql.checkData(0, 0, 0)
tdSql.query('select count(c_1), min(c_1),tbname from d2.can partition by tbname', queryTimes=1)
tdSql.query('select count(c_1), min(c_1),tbname from d2.can partition by tbname order by 3', queryTimes=1)
tdSql.checkData(0, 0, 0)
tdSql.checkData(0, 1, None)
tdSql.checkData(0, 2, 't3')
tdSql.checkData(0, 2, 't1')
tdSql.checkData(1, 0, 15)
tdSql.checkData(1, 1, 1471617148940980000)
@ -268,7 +268,7 @@ class TDTestCase:
tdSql.checkData(2, 0, 0)
tdSql.checkData(2, 1, None)
tdSql.checkData(2, 2, 't1')
tdSql.checkData(2, 2, 't3')
def run(self):
self.test_count_with_sma_data()

View File

@ -982,6 +982,9 @@ int smlProcess_18784_Test() {
rowIndex++;
}
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists db_18784");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1187,6 +1190,9 @@ int sml_ts2164_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists line_test");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1236,6 +1242,8 @@ int sml_ts3116_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
code = taos_errno(pRes);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists ts3116");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1449,6 +1457,13 @@ int sml_td24070_Test() {
code = taos_errno(pRes);
ASSERT(code == 0);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td24070_write");
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td24070_read");
taos_free_result(pRes);
taos_close(taos);
// test table privilege
@ -1477,6 +1492,10 @@ int sml_td23881_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists line_23881");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1691,6 +1710,8 @@ int sml_ts2385_Test() {
code = taos_errno(pRes);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists ts2385");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1784,6 +1805,9 @@ int sml_td24559_Test() {
}
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td24559");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1889,6 +1913,9 @@ int sml_td29691_Test() {
ASSERT(code == TSDB_CODE_PAR_DUPLICATED_COLUMN);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td29691");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -1942,6 +1969,9 @@ int sml_td18789_Test() {
}
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td18789");
taos_free_result(pRes);
taos_close(taos);
return code;
@ -2047,6 +2077,9 @@ int sml_td29373_Test() {
ASSERT(code == TSDB_CODE_SML_NOT_SUPPORT_PK);
taos_free_result(pRes);
pRes = taos_query(taos, "drop database if exists td29373");
taos_free_result(pRes);
taos_close(taos);
return code;