xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch '3.0' into fix/TD-27216-3

This commit is contained in:
Yihao Deng 2023-11-09 17:12:56 +08:00 committed by GitHub
parent 15b73354f8 9dae2250c3
commit 0453991cdf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
49 changed files with 1665 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/vnode/src/inc/vndCos.h → include/common/cos.h
View File

@ -16,7 +16,7 @@
#ifndef _TD_VND_COS_H_
#define _TD_VND_COS_H_
#include "vnd.h"
#include "os.h"
#ifdef __cplusplus
extern "C" {
@ -24,6 +24,7 @@ extern "C" {
#define S3_BLOCK_CACHE
extern int8_t tsS3StreamEnabled;
extern int8_t tsS3Enabled;
extern int32_t tsS3BlockSize;
extern int32_t tsS3BlockCacheSize;
@ -38,6 +39,7 @@ void s3DeleteObjectsByPrefix(const char *prefix);
void s3DeleteObjects(const char *object_name[], int nobject);
bool s3Exists(const char *object_name);
bool s3Get(const char *object_name, const char *path);
int32_t s3GetObjectsByPrefix(const char *prefix, const char* path);
int32_t s3GetObjectBlock(const char *object_name, int64_t offset, int64_t size, uint8_t **ppBlock);
void s3EvictCache(const char *path, long object_size);
long s3Size(const char *object_name);

24
include/common/rsync.h Normal file
View File

@ -0,0 +1,24 @@
//
// Created by mingming wanng on 2023/11/2.
//
#ifndef TDENGINE_RSYNC_H
#define TDENGINE_RSYNC_H
#ifdef __cplusplus
extern "C" {
#endif
#include "tarray.h"
void stopRsync();
void startRsync();
int uploadRsync(char* id, char* path);
int downloadRsync(char* id, char* path);
int deleteRsync(char* id);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_RSYNC_H

1
include/common/tcommon.h
View File

@ -249,6 +249,7 @@ typedef struct SQueryTableDataCond {
SColumnInfo* colList;
int32_t* pSlotList; // the column output destation slot, and it may be null
int32_t type; // data block load type:
bool skipRollup;
STimeWindow twindows;
int64_t startVersion;
int64_t endVersion;

7
include/common/tglobal.h
View File

@ -75,6 +75,13 @@ extern int32_t tsElectInterval;
extern int32_t tsHeartbeatInterval;
extern int32_t tsHeartbeatTimeout;
// snode
extern int32_t tsRsyncPort;
extern char tsCheckpointBackupDir[];
// vnode checkpoint
extern char tsSnodeAddress[]; //127.0.0.1:873
// mnode
extern int64_t tsMndSdbWriteDelta;
extern int64_t tsMndLogRetention;

1
include/libs/executor/executor.h
View File

@ -49,6 +49,7 @@ typedef struct {
uint64_t checkpointId;
bool initTableReader;
bool initTqReader;
bool skipRollup;
int32_t numOfVgroups;
void* sContext; // SSnapContext*
void* pStateBackend;

1
include/libs/function/function.h
View File

@ -168,7 +168,6 @@ typedef struct {
struct SStreamFileState *pFileState;
int32_t number;
SSHashObj *parNameMap;
int64_t checkPointId;
int32_t taskId;
int64_t streamId;
int64_t streamBackendRid;

2
include/libs/stream/tstream.h
View File

@ -826,6 +826,7 @@ void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
int32_t streamMetaReopen(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta);
int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta);
int64_t streamMetaGetLatestCheckpointId(SStreamMeta* pMeta);
void streamMetaNotifyClose(SStreamMeta* pMeta);
void streamMetaStartHb(SStreamMeta* pMeta);
void streamMetaInitForSnode(SStreamMeta* pMeta);
@ -840,6 +841,7 @@ void streamMetaResetStartInfo(STaskStartInfo* pMeta);
// checkpoint
int32_t streamProcessCheckpointSourceReq(SStreamTask* pTask, SStreamCheckpointSourceReq* pReq);
int32_t streamProcessCheckpointReadyMsg(SStreamTask* pTask);
int32_t streamTaskBuildCheckpoint(SStreamTask* pTask);
void streamTaskClearCheckInfo(SStreamTask* pTask);
int32_t streamAlignTransferState(SStreamTask* pTask);
int32_t streamBuildAndSendDropTaskMsg(SMsgCb* pMsgCb, int32_t vgId, SStreamTaskId* pTaskId);

6
include/os/osDef.h
View File

@ -237,6 +237,12 @@ void syslog(int unused, const char *format, ...);
#define TD_DIRSEP "/"
#endif
#if defined(_WIN32)
#define TD_DIRSEP_CHAR '\\'
#else
#define TD_DIRSEP_CHAR '/'
#endif
#define TD_LOCALE_LEN 64
#define TD_CHARSET_LEN 64
#define TD_TIMEZONE_LEN 96

69
source/common/CMakeLists.txt
View File

@ -46,6 +46,75 @@ target_link_libraries(
INTERFACE api
)
if(${BUILD_S3})
if(${BUILD_WITH_S3})
target_include_directories(
common
PUBLIC "$ENV{HOME}/.cos-local.2/include"
)
set(CMAKE_FIND_LIBRARY_SUFFIXES ".a")
set(CMAKE_PREFIX_PATH $ENV{HOME}/.cos-local.2)
find_library(S3_LIBRARY s3)
find_library(CURL_LIBRARY curl $ENV{HOME}/.cos-local.2/lib NO_DEFAULT_PATH)
find_library(XML2_LIBRARY xml2)
find_library(SSL_LIBRARY ssl $ENV{HOME}/.cos-local.2/lib64 NO_DEFAULT_PATH)
find_library(CRYPTO_LIBRARY crypto $ENV{HOME}/.cos-local.2/lib64 NO_DEFAULT_PATH)
target_link_libraries(
common
# s3
PUBLIC ${S3_LIBRARY}
PUBLIC ${CURL_LIBRARY}
PUBLIC ${SSL_LIBRARY}
PUBLIC ${CRYPTO_LIBRARY}
PUBLIC ${XML2_LIBRARY}
)
add_definitions(-DUSE_S3)
endif()
if(${BUILD_WITH_COS})
set(CMAKE_FIND_LIBRARY_SUFFIXES ".a")
find_library(APR_LIBRARY apr-1 PATHS /usr/local/apr/lib/)
find_library(APR_UTIL_LIBRARY aprutil-1 PATHS /usr/local/apr/lib/)
find_library(MINIXML_LIBRARY mxml)
find_library(CURL_LIBRARY curl)
target_link_libraries(
common
# s3
PUBLIC cos_c_sdk_static
PUBLIC ${APR_UTIL_LIBRARY}
PUBLIC ${APR_LIBRARY}
PUBLIC ${MINIXML_LIBRARY}
PUBLIC ${CURL_LIBRARY}
)
# s3
FIND_PROGRAM(APR_CONFIG_BIN NAMES apr-config apr-1-config PATHS /usr/bin /usr/local/bin /usr/local/apr/bin/)
IF (APR_CONFIG_BIN)
EXECUTE_PROCESS(
COMMAND ${APR_CONFIG_BIN} --includedir
OUTPUT_VARIABLE APR_INCLUDE_DIR
OUTPUT_STRIP_TRAILING_WHITESPACE
)
ENDIF()
include_directories (${APR_INCLUDE_DIR})
target_include_directories(
common
PUBLIC "${TD_SOURCE_DIR}/contrib/cos-c-sdk-v5/cos_c_sdk"
PUBLIC "$ENV{HOME}/.cos-local.1/include"
)
add_definitions(-DUSE_COS)
endif(${BUILD_WITH_COS})
endif()
if(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
endif(${BUILD_TEST})

126
source/dnode/vnode/src/vnd/vnodeCos.c → source/common/src/cos.c
View File

@ -1,6 +1,6 @@
#define ALLOW_FORBID_FUNC
#include "vndCos.h"
#include "cos.h"
extern char tsS3Endpoint[];
extern char tsS3AccessKeyId[];
@ -13,6 +13,7 @@ extern int8_t tsS3Https;
#if defined(USE_S3)
#include "libs3.h"
#include "tarray.h"
static int verifyPeerG = 0;
static const char *awsRegionG = NULL;
@ -34,7 +35,7 @@ static int32_t s3Begin() {
}
if ((status = S3_initialize("s3", verifyPeerG | S3_INIT_ALL, hostname)) != S3StatusOK) {
vError("Failed to initialize libs3: %s\n", S3_get_status_name(status));
uError("Failed to initialize libs3: %s\n", S3_get_status_name(status));
return -1;
}
@ -65,12 +66,18 @@ static int should_retry() {
static void s3PrintError(const char *func, S3Status status, char error_details[]) {
if (status < S3StatusErrorAccessDenied) {
vError("%s: %s", __func__, S3_get_status_name(status));
uError("%s: %s", __func__, S3_get_status_name(status));
} else {
vError("%s: %s, %s", __func__, S3_get_status_name(status), error_details);
uError("%s: %s, %s", __func__, S3_get_status_name(status), error_details);
}
}
typedef struct {
char err_msg[128];
S3Status status;
TdFilePtr file;
} TS3GetData;
typedef struct {
char err_msg[128];
S3Status status;
@ -78,6 +85,11 @@ typedef struct {
char *buf;
} TS3SizeCBD;
static S3Status responsePropertiesCallbackNull(const S3ResponseProperties *properties, void *callbackData) {
// (void)callbackData;
return S3StatusOK;
}
static S3Status responsePropertiesCallback(const S3ResponseProperties *properties, void *callbackData) {
//(void)callbackData;
TS3SizeCBD *cbd = callbackData;
@ -291,7 +303,7 @@ S3Status initial_multipart_callback(const char *upload_id, void *callbackData) {
}
S3Status MultipartResponseProperiesCallback(const S3ResponseProperties *properties, void *callbackData) {
responsePropertiesCallback(properties, callbackData);
responsePropertiesCallbackNull(properties, callbackData);
MultipartPartData *data = (MultipartPartData *)callbackData;
int seq = data->seq;
@ -390,7 +402,7 @@ static int try_get_parts_info(const char *bucketName, const char *key, UploadMan
S3BucketContext bucketContext = {0, tsS3BucketName, protocolG, uriStyleG, tsS3AccessKeyId, tsS3AccessKeySecret,
0, awsRegionG};
S3ListPartsHandler listPartsHandler = {{&responsePropertiesCallback, &responseCompleteCallback}, &listPartsCallback};
S3ListPartsHandler listPartsHandler = {{&responsePropertiesCallbackNull, &responseCompleteCallback}, &listPartsCallback};
list_parts_callback_data data;
@ -445,13 +457,13 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
data.noStatus = noStatus;
if (taosStatFile(file, &contentLength, NULL, NULL) < 0) {
vError("ERROR: %s Failed to stat file %s: ", __func__, file);
uError("ERROR: %s Failed to stat file %s: ", __func__, file);
code = TAOS_SYSTEM_ERROR(errno);
return code;
}
if (!(data.infileFD = taosOpenFile(file, TD_FILE_READ))) {
vError("ERROR: %s Failed to open file %s: ", __func__, file);
uError("ERROR: %s Failed to open file %s: ", __func__, file);
code = TAOS_SYSTEM_ERROR(errno);
return code;
}
@ -469,7 +481,7 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
metaProperties, useServerSideEncryption};
if (contentLength <= MULTIPART_CHUNK_SIZE) {
S3PutObjectHandler putObjectHandler = {{&responsePropertiesCallback, &responseCompleteCallback},
S3PutObjectHandler putObjectHandler = {{&responsePropertiesCallbackNull, &responseCompleteCallback},
&putObjectDataCallback};
do {
@ -486,7 +498,7 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
s3PrintError(__func__, data.status, data.err_msg);
code = TAOS_SYSTEM_ERROR(EIO);
} else if (data.contentLength) {
vError("ERROR: %s Failed to read remaining %llu bytes from input", __func__,
uError("ERROR: %s Failed to read remaining %llu bytes from input", __func__,
(unsigned long long)data.contentLength);
code = TAOS_SYSTEM_ERROR(EIO);
}
@ -506,14 +518,14 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
memset(&partData, 0, sizeof(MultipartPartData));
int partContentLength = 0;
S3MultipartInitialHandler handler = {{&responsePropertiesCallback, &responseCompleteCallback},
S3MultipartInitialHandler handler = {{&responsePropertiesCallbackNull, &responseCompleteCallback},
&initial_multipart_callback};
S3PutObjectHandler putObjectHandler = {{&MultipartResponseProperiesCallback, &responseCompleteCallback},
&putObjectDataCallback};
S3MultipartCommitHandler commit_handler = {
{&responsePropertiesCallback, &responseCompleteCallback}, &multipartPutXmlCallback, 0};
{&responsePropertiesCallbackNull, &responseCompleteCallback}, &multipartPutXmlCallback, 0};
manager.etags = (char **)taosMemoryMalloc(sizeof(char *) * totalSeq);
manager.next_etags_pos = 0;
@ -658,19 +670,19 @@ static void s3FreeObjectKey(void *pItem) {
taosMemoryFree(key);
}
void s3DeleteObjectsByPrefix(const char *prefix) {
static SArray* getListByPrefix(const char *prefix){
S3BucketContext bucketContext = {0, tsS3BucketName, protocolG, uriStyleG, tsS3AccessKeyId, tsS3AccessKeySecret,
0, awsRegionG};
S3ListBucketHandler listBucketHandler = {{&responsePropertiesCallback, &responseCompleteCallback},
S3ListBucketHandler listBucketHandler = {{&responsePropertiesCallbackNull, &responseCompleteCallback},
&listBucketCallback};
const char *marker = 0, *delimiter = 0;
int maxkeys = 0, allDetails = 0;
list_bucket_callback_data data;
data.objectArray = taosArrayInit(32, POINTER_BYTES);
data.objectArray = taosArrayInit(32, sizeof(void*));
if (!data.objectArray) {
vError("%s: %s", __func__, "out of memoty");
return;
uError("%s: %s", __func__, "out of memoty");
return NULL;
}
if (marker) {
snprintf(data.nextMarker, sizeof(data.nextMarker), "%s", marker);
@ -693,18 +705,15 @@ void s3DeleteObjectsByPrefix(const char *prefix) {
if (data.status == S3StatusOK) {
if (data.keyCount > 0) {
// printListBucketHeader(allDetails);
s3DeleteObjects(TARRAY_DATA(data.objectArray), TARRAY_SIZE(data.objectArray));
return data.objectArray;
}
} else {
s3PrintError(__func__, data.status, data.err_msg);
}
taosArrayDestroyEx(data.objectArray, s3FreeObjectKey);
return NULL;
}
void s3DeleteObjects(const char *object_name[], int nobject) {
int status = 0;
S3BucketContext bucketContext = {0, tsS3BucketName, protocolG, uriStyleG, tsS3AccessKeyId, tsS3AccessKeySecret,
0, awsRegionG};
S3ResponseHandler responseHandler = {0, &responseCompleteCallback};
@ -721,6 +730,13 @@ void s3DeleteObjects(const char *object_name[], int nobject) {
}
}
void s3DeleteObjectsByPrefix(const char *prefix) {
SArray* objectArray = getListByPrefix(prefix);
if(objectArray == NULL)return;
s3DeleteObjects(TARRAY_DATA(objectArray), TARRAY_SIZE(objectArray));
taosArrayDestroyEx(objectArray, s3FreeObjectKey);
}
static S3Status getObjectDataCallback(int bufferSize, const char *buffer, void *callbackData) {
TS3SizeCBD *cbd = callbackData;
if (cbd->content_length != bufferSize) {
@ -758,7 +774,7 @@ int32_t s3GetObjectBlock(const char *object_name, int64_t offset, int64_t size,
} while (S3_status_is_retryable(cbd.status) && should_retry());
if (cbd.status != S3StatusOK) {
vError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
uError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
return TAOS_SYSTEM_ERROR(EIO);
}
@ -767,6 +783,67 @@ int32_t s3GetObjectBlock(const char *object_name, int64_t offset, int64_t size,
return 0;
}
static S3Status getObjectCallback(int bufferSize, const char *buffer, void *callbackData) {
TS3GetData *cbd = (TS3GetData *) callbackData;
size_t wrote = taosWriteFile(cbd->file, buffer, bufferSize);
return ((wrote < (size_t) bufferSize) ?
S3StatusAbortedByCallback : S3StatusOK);
}
int32_t s3GetObjectToFile(const char *object_name, char* fileName) {
int64_t ifModifiedSince = -1, ifNotModifiedSince = -1;
const char *ifMatch = 0, *ifNotMatch = 0;
S3BucketContext bucketContext = {0, tsS3BucketName, protocolG, uriStyleG, tsS3AccessKeyId, tsS3AccessKeySecret,
0, awsRegionG};
S3GetConditions getConditions = {ifModifiedSince, ifNotModifiedSince, ifMatch, ifNotMatch};
S3GetObjectHandler getObjectHandler = {{&responsePropertiesCallbackNull, &responseCompleteCallback},
&getObjectCallback};
TdFilePtr pFile = taosOpenFile(fileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
uError("[s3] open file error, errno:%d, fileName:%s", errno, fileName);
return -1;
}
TS3GetData cbd = {0};
cbd.file = pFile;
do {
S3_get_object(&bucketContext, object_name, &getConditions, 0, 0, 0, 0, &getObjectHandler, &cbd);
} while (S3_status_is_retryable(cbd.status) && should_retry());
if (cbd.status != S3StatusOK) {
uError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
taosCloseFile(&pFile);
return TAOS_SYSTEM_ERROR(EIO);
}
taosCloseFile(&pFile);
return 0;
}
int32_t s3GetObjectsByPrefix(const char *prefix, const char* path){
SArray* objectArray = getListByPrefix(prefix);
if(objectArray == NULL) return -1;
for(size_t i = 0; i < taosArrayGetSize(objectArray); i++){
char* object = taosArrayGetP(objectArray, i);
const char* tmp = strchr(object, '/');
tmp = (tmp == NULL) ? object : tmp + 1;
char fileName[PATH_MAX] = {0};
if(path[strlen(path) - 1] != TD_DIRSEP_CHAR){
snprintf(fileName, PATH_MAX, "%s%s%s", path, TD_DIRSEP, tmp);
}else{
snprintf(fileName, PATH_MAX, "%s%s", path, tmp);
}
if(s3GetObjectToFile(object, fileName) != 0){
taosArrayDestroyEx(objectArray, s3FreeObjectKey);
return -1;
}
}
taosArrayDestroyEx(objectArray, s3FreeObjectKey);
return 0;
}
long s3Size(const char *object_name) {
long size = 0;
int status = 0;
@ -782,7 +859,7 @@ long s3Size(const char *object_name) {
} while (S3_status_is_retryable(cbd.status) && should_retry());
if ((cbd.status != S3StatusOK) && (cbd.status != S3StatusErrorPreconditionFailed)) {
vError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
uError("%s: %d(%s)", __func__, cbd.status, cbd.err_msg);
}
size = cbd.content_length;
@ -1237,6 +1314,7 @@ void s3DeleteObjectsByPrefix(const char *prefix) {}
void s3DeleteObjects(const char *object_name[], int nobject) {}
bool s3Exists(const char *object_name) { return false; }
bool s3Get(const char *object_name, const char *path) { return false; }
int32_t s3GetObjectsByPrefix(const char *prefix, const char* path) { return 0; }
int32_t s3GetObjectBlock(const char *object_name, int64_t offset, int64_t size, uint8_t **ppBlock) { return 0; }
void s3EvictCache(const char *path, long object_size) {}
long s3Size(const char *object_name) { return 0; }

235
source/common/src/rsync.c Normal file
View File

@ -0,0 +1,235 @@
//
// Created by mingming wanng on 2023/11/2.
//
#include "rsync.h"
#include <stdlib.h>
#include "tglobal.h"
#define ERRNO_ERR_FORMAT "errno:%d,msg:%s"
#define ERRNO_ERR_DATA errno,strerror(errno)
// deleteRsync function produce empty directories, traverse base directory to remove them
static void removeEmptyDir(){
TdDirPtr pDir = taosOpenDir(tsCheckpointBackupDir);
if (pDir == NULL) return;
TdDirEntryPtr de = NULL;
while ((de = taosReadDir(pDir)) != NULL) {
if (!taosDirEntryIsDir(de)) {
continue;
}
if (strcmp(taosGetDirEntryName(de), ".") == 0 || strcmp(taosGetDirEntryName(de), "..") == 0) continue;
char filename[PATH_MAX] = {0};
snprintf(filename, sizeof(filename), "%s%s", tsCheckpointBackupDir, taosGetDirEntryName(de));
TdDirPtr pDirTmp = taosOpenDir(filename);
TdDirEntryPtr deTmp = NULL;
bool empty = true;
while ((deTmp = taosReadDir(pDirTmp)) != NULL){
if (strcmp(taosGetDirEntryName(deTmp), ".") == 0 || strcmp(taosGetDirEntryName(deTmp), "..") == 0) continue;
empty = false;
}
if(empty) taosRemoveDir(filename);
taosCloseDir(&pDirTmp);
}
taosCloseDir(&pDir);
}
#ifdef WINDOWS
// C:\TDengine\data\backup\checkpoint\ -> /c/TDengine/data/backup/checkpoint/
static void changeDirFromWindowsToLinux(char* from, char* to){
to[0] = '/';
to[1] = from[0];
for(int i = 2; i < strlen(from); i++) {
if (from[i] == '\\') {
to[i] = '/';
} else {
to[i] = from[i];
}
}
}
#endif
static int generateConfigFile(char* confDir){
TdFilePtr pFile = taosOpenFile(confDir, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
uError("[rsync] open conf file error, dir:%s,"ERRNO_ERR_FORMAT, confDir, ERRNO_ERR_DATA);
return -1;
}
#ifdef WINDOWS
char path[PATH_MAX] = {0};
changeDirFromWindowsToLinux(tsCheckpointBackupDir, path);
#endif
char confContent[PATH_MAX*4] = {0};
snprintf(confContent, PATH_MAX*4,
#ifndef WINDOWS
"uid = root\n"
"gid = root\n"
#endif
"use chroot = false\n"
"max connections = 200\n"
"timeout = 100\n"
"lock file = %srsync.lock\n"
"log file = %srsync.log\n"
"ignore errors = true\n"
"read only = false\n"
"list = false\n"
"[checkpoint]\n"
"path = %s", tsCheckpointBackupDir, tsCheckpointBackupDir,
#ifdef WINDOWS
path
#else
tsCheckpointBackupDir
#endif
);
uDebug("[rsync] conf:%s", confContent);
if (taosWriteFile(pFile, confContent, strlen(confContent)) <= 0){
uError("[rsync] write conf file error,"ERRNO_ERR_FORMAT, ERRNO_ERR_DATA);
taosCloseFile(&pFile);
return -1;
}
taosCloseFile(&pFile);
return 0;
}
static int execCommand(char* command){
int try = 3;
int32_t code = 0;
while(try-- > 0) {
code = system(command);
if (code == 0) {
break;
}
taosMsleep(10);
}
return code;
}
void stopRsync(){
int code =
#ifdef WINDOWS
system("taskkill /f /im rsync.exe");
#else
system("pkill rsync");
#endif
if(code != 0){
uError("[rsync] stop rsync server failed,"ERRNO_ERR_FORMAT, ERRNO_ERR_DATA);
return;
}
uDebug("[rsync] stop rsync server successful");
}
void startRsync(){
if(taosMulMkDir(tsCheckpointBackupDir) != 0){
uError("[rsync] build checkpoint backup dir failed, dir:%s,"ERRNO_ERR_FORMAT, tsCheckpointBackupDir, ERRNO_ERR_DATA);
return;
}
removeEmptyDir();
char confDir[PATH_MAX] = {0};
snprintf(confDir, PATH_MAX, "%srsync.conf", tsCheckpointBackupDir);
int code = generateConfigFile(confDir);
if(code != 0){
return;
}
char cmd[PATH_MAX] = {0};
snprintf(cmd, PATH_MAX, "rsync --daemon --port=%d --config=%s", tsRsyncPort, confDir);
// start rsync service to backup checkpoint
code = system(cmd);
if(code != 0){
uError("[rsync] start server failed, code:%d,"ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return;
}
uDebug("[rsync] start server successful");
}
int uploadRsync(char* id, char* path){
#ifdef WINDOWS
char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform);
#endif
char command[PATH_MAX] = {0};
#ifdef WINDOWS
if(pathTransform[strlen(pathTransform) - 1] != '/'){
#else
if(path[strlen(path) - 1] != '/'){
#endif
snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 --bwlimit=100000 %s/ rsync://%s/checkpoint/%s/",
#ifdef WINDOWS
pathTransform
#else
path
#endif
, tsSnodeAddress, id);
}else{
snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 --bwlimit=100000 %s rsync://%s/checkpoint/%s/",
#ifdef WINDOWS
pathTransform
#else
path
#endif
, tsSnodeAddress, id);
}
int code = execCommand(command);
if(code != 0){
uError("[rsync] send failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return -1;
}
uDebug("[rsync] upload data:%s successful", id);
return 0;
}
int downloadRsync(char* id, char* path){
#ifdef WINDOWS
char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform);
#endif
char command[PATH_MAX] = {0};
snprintf(command, PATH_MAX, "rsync -av --timeout=10 --bwlimit=100000 rsync://%s/checkpoint/%s/ %s",
tsSnodeAddress, id,
#ifdef WINDOWS
pathTransform
#else
path
#endif
);
int code = execCommand(command);
if(code != 0){
uError("[rsync] get failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return -1;
}
uDebug("[rsync] down data:%s successful", id);
return 0;
}
int deleteRsync(char* id){
char* tmp = "./tmp_empty/";
int code = taosMkDir(tmp);
if(code != 0){
uError("[rsync] make tmp dir failed. code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return -1;
}
char command[PATH_MAX] = {0};
snprintf(command, PATH_MAX, "rsync -av --delete --timeout=10 %s rsync://%s/checkpoint/%s/",
tmp, tsSnodeAddress, id);
code = execCommand(command);
taosRemoveDir(tmp);
if(code != 0){
uError("[rsync] get failed code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return -1;
}
uDebug("[rsync] delete data:%s successful", id);
return 0;
}

24
source/common/src/tglobal.c
View File

@ -127,6 +127,15 @@ char tsSmlAutoChildTableNameDelimiter[TSDB_TABLE_NAME_LEN] = "";
// bool tsSmlDataFormat = false;
// int32_t tsSmlBatchSize = 10000;
// checkpoint backup
char tsSnodeAddress[TSDB_FQDN_LEN] = {0};
int32_t tsRsyncPort = 873;
#ifdef WINDOWS
char tsCheckpointBackupDir[PATH_MAX] = "C:\\TDengine\\data\\backup\\checkpoint\\";
#else
char tsCheckpointBackupDir[PATH_MAX] = "/var/lib/taos/backup/checkpoint/";
#endif
// tmq
int32_t tmqMaxTopicNum = 20;
// query
@ -260,6 +269,7 @@ char tsS3AccessKeySecret[TSDB_FQDN_LEN] = "<accesskeysecrect>";
char tsS3BucketName[TSDB_FQDN_LEN] = "<bucketname>";
char tsS3AppId[TSDB_FQDN_LEN] = "<appid>";
int8_t tsS3Enabled = false;
int8_t tsS3StreamEnabled = false;
int8_t tsS3Https = true;
char tsS3Hostname[TSDB_FQDN_LEN] = "<hostname>";
@ -323,9 +333,10 @@ int32_t taosSetS3Cfg(SConfig *pCfg) {
tstrncpy(tsS3AppId, appid + 1, TSDB_FQDN_LEN);
}
}
if (tsS3BucketName[0] != '<' && tsDiskCfgNum > 1) {
if (tsS3BucketName[0] != '<') {
#if defined(USE_COS) || defined(USE_S3)
tsS3Enabled = true;
if(tsDiskCfgNum > 1) tsS3Enabled = true;
tsS3StreamEnabled = true;
#endif
}
@ -594,7 +605,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
0)
return -1;
tsNumOfVnodeRsmaThreads = tsNumOfCores;
tsNumOfVnodeRsmaThreads = tsNumOfCores / 2;
tsNumOfVnodeRsmaThreads = TMAX(tsNumOfVnodeRsmaThreads, 4);
if (cfgAddInt32(pCfg, "numOfVnodeRsmaThreads", tsNumOfVnodeRsmaThreads, 1, 1024, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0)
return -1;
@ -661,6 +672,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddString(pCfg, "telemetryServer", tsTelemServer, CFG_SCOPE_BOTH, CFG_DYN_BOTH) != 0) return -1;
if (cfgAddInt32(pCfg, "telemetryPort", tsTelemPort, 1, 65056, CFG_SCOPE_BOTH, CFG_DYN_NONE) != 0) return -1;
if (cfgAddInt32(pCfg, "rsyncPort", tsRsyncPort, 1, 65535, CFG_SCOPE_BOTH, CFG_DYN_SERVER) != 0) return -1;
if (cfgAddString(pCfg, "snodeAddress", tsSnodeAddress, CFG_SCOPE_SERVER, CFG_DYN_SERVER) != 0) return -1;
if (cfgAddString(pCfg, "checkpointBackupDir", tsCheckpointBackupDir, CFG_SCOPE_SERVER, CFG_DYN_SERVER) != 0) return -1;
if (cfgAddInt32(pCfg, "tmqMaxTopicNum", tmqMaxTopicNum, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0)
return -1;
@ -1099,7 +1114,10 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsTtlChangeOnWrite = cfgGetItem(pCfg, "ttlChangeOnWrite")->bval;
tsTtlFlushThreshold = cfgGetItem(pCfg, "ttlFlushThreshold")->i32;
tsTelemInterval = cfgGetItem(pCfg, "telemetryInterval")->i32;
tsRsyncPort = cfgGetItem(pCfg, "rsyncPort")->i32;
tstrncpy(tsTelemServer, cfgGetItem(pCfg, "telemetryServer")->str, TSDB_FQDN_LEN);
tstrncpy(tsSnodeAddress, cfgGetItem(pCfg, "snodeAddress")->str, TSDB_FQDN_LEN);
tstrncpy(tsCheckpointBackupDir, cfgGetItem(pCfg, "checkpointBackupDir")->str, PATH_MAX);
tsTelemPort = (uint16_t)cfgGetItem(pCfg, "telemetryPort")->i32;
tmqMaxTopicNum = cfgGetItem(pCfg, "tmqMaxTopicNum")->i32;

7
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -144,6 +144,10 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
char path[TSDB_FILENAME_LEN] = {0};
bool atExit = true;
if (pVnode->failed) {
ASSERT(pVnode->pImpl == NULL);
goto _closed;
}
if (vnodeIsLeader(pVnode->pImpl)) {
vnodeProposeCommitOnNeed(pVnode->pImpl, atExit);
}
@ -202,6 +206,8 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal)
vnodeClose(pVnode->pImpl);
pVnode->pImpl = NULL;
_closed:
dInfo("vgId:%d, vnode is closed", pVnode->vgId);
if (commitAndRemoveWal) {
@ -386,7 +392,6 @@ static void *vmCloseVnodeInThread(void *param) {
for (int32_t v = 0; v < pThread->vnodeNum; ++v) {
SVnodeObj *pVnode = pThread->ppVnodes[v];
if (pVnode->failed) continue;
char stepDesc[TSDB_STEP_DESC_LEN] = {0};
snprintf(stepDesc, TSDB_STEP_DESC_LEN, "vgId:%d, start to close, %d of %d have been closed", pVnode->vgId,

4
source/dnode/snode/src/snode.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "rsync.h"
#include "executor.h"
#include "sndInt.h"
#include "tstream.h"
@ -122,6 +123,9 @@ SSnode *sndOpen(const char *path, const SSnodeOpt *pOption) {
goto FAIL;
}
stopRsync();
startRsync();
// todo fix it: send msg to mnode to rollback to an existed checkpoint
streamMetaInitForSnode(pSnode->pMeta);
return pSnode;

70
source/dnode/vnode/CMakeLists.txt
View File

@ -8,7 +8,6 @@ set(
"src/vnd/vnodeCommit.c"
"src/vnd/vnodeQuery.c"
"src/vnd/vnodeModule.c"
"src/vnd/vnodeCos.c"
"src/vnd/vnodeSvr.c"
"src/vnd/vnodeSync.c"
"src/vnd/vnodeSnapshot.c"
@ -161,75 +160,6 @@ target_link_libraries(
PUBLIC index
)
if(${BUILD_S3})
if(${BUILD_WITH_S3})
target_include_directories(
vnode
PUBLIC "$ENV{HOME}/.cos-local.2/include"
)
set(CMAKE_FIND_LIBRARY_SUFFIXES ".a")
set(CMAKE_PREFIX_PATH $ENV{HOME}/.cos-local.2)
find_library(S3_LIBRARY s3)
find_library(CURL_LIBRARY curl $ENV{HOME}/.cos-local.2/lib NO_DEFAULT_PATH)
find_library(XML2_LIBRARY xml2)
find_library(SSL_LIBRARY ssl $ENV{HOME}/.cos-local.2/lib $ENV{HOME}/.cos-local.2/lib64 NO_DEFAULT_PATH)
find_library(CRYPTO_LIBRARY crypto $ENV{HOME}/.cos-local.2/lib $ENV{HOME}/.cos-local.2/lib64 NO_DEFAULT_PATH)
target_link_libraries(
vnode
# s3
PUBLIC ${S3_LIBRARY}
PUBLIC ${CURL_LIBRARY}
PUBLIC ${SSL_LIBRARY}
PUBLIC ${CRYPTO_LIBRARY}
PUBLIC ${XML2_LIBRARY}
)
add_definitions(-DUSE_S3)
endif()
if(${BUILD_WITH_COS})
set(CMAKE_FIND_LIBRARY_SUFFIXES ".a")
find_library(APR_LIBRARY apr-1 PATHS /usr/local/apr/lib/)
find_library(APR_UTIL_LIBRARY aprutil-1 PATHS /usr/local/apr/lib/)
find_library(MINIXML_LIBRARY mxml)
find_library(CURL_LIBRARY curl)
target_link_libraries(
vnode
# s3
PUBLIC cos_c_sdk_static
PUBLIC ${APR_UTIL_LIBRARY}
PUBLIC ${APR_LIBRARY}
PUBLIC ${MINIXML_LIBRARY}
PUBLIC ${CURL_LIBRARY}
)
# s3
FIND_PROGRAM(APR_CONFIG_BIN NAMES apr-config apr-1-config PATHS /usr/bin /usr/local/bin /usr/local/apr/bin/)
IF (APR_CONFIG_BIN)
EXECUTE_PROCESS(
COMMAND ${APR_CONFIG_BIN} --includedir
OUTPUT_VARIABLE APR_INCLUDE_DIR
OUTPUT_STRIP_TRAILING_WHITESPACE
)
ENDIF()
include_directories (${APR_INCLUDE_DIR})
target_include_directories(
vnode
PUBLIC "${TD_SOURCE_DIR}/contrib/cos-c-sdk-v5/cos_c_sdk"
PUBLIC "$ENV{HOME}/.cos-local.1/include"
)
add_definitions(-DUSE_COS)
endif(${BUILD_WITH_COS})
endif()
IF (TD_GRANT)
TARGET_LINK_LIBRARIES(vnode PUBLIC grant)
ENDIF ()

35
source/dnode/vnode/src/inc/sma.h
View File

@ -108,6 +108,7 @@ struct SRSmaStat {
int64_t refId; // shared by fetch tasks
volatile int64_t nBufItems; // number of items in queue buffer
SRWLatch lock; // r/w lock for rsma fs(e.g. qtaskinfo)
volatile int32_t execStat; // 0 succeed, other failed
volatile int32_t nFetchAll; // active number of fetch all
volatile int8_t triggerStat; // shared by fetch tasks
volatile int8_t commitStat; // 0 not in committing, 1 in committing
@ -115,6 +116,7 @@ struct SRSmaStat {
SRSmaFS fs; // for recovery/snapshot r/w
SHashObj *infoHash; // key: suid, value: SRSmaInfo
tsem_t notEmpty; // has items in queue buffer
SArray *blocks; // SArray<SSDataBlock>
};
struct SSmaStat {
@ -136,13 +138,18 @@ struct SSmaStat {
#define RSMA_FS_LOCK(r) (&(r)->lock)
struct SRSmaInfoItem {
int8_t level : 4;
int8_t fetchLevel : 4;
int8_t triggerStat;
uint16_t nScanned;
int32_t maxDelay; // ms
tmr_h tmrId;
void *pStreamState;
int8_t level;
int8_t fetchLevel;
int8_t triggerStat;
int32_t nScanned;
int32_t streamFlushed : 1;
int32_t maxDelay : 31; // ms
int64_t submitReqVer;
int64_t fetchResultVer;
tmr_h tmrId;
void *pStreamState;
void *pStreamTask; // SStreamTask
SArray *pResList;
};
struct SRSmaInfo {
@ -160,12 +167,10 @@ struct SRSmaInfo {
STaosQall *qall; // buffer qall of SubmitReq
};
#define RSMA_INFO_HEAD_LEN offsetof(SRSmaInfo, items)
#define RSMA_INFO_IS_DEL(r) ((r)->delFlag == 1)
#define RSMA_INFO_SET_DEL(r) ((r)->delFlag = 1)
#define RSMA_INFO_QTASK(r, i) ((r)->taskInfo[i])
#define RSMA_INFO_IQTASK(r, i) ((r)->iTaskInfo[i])
#define RSMA_INFO_ITEM(r, i) (&(r)->items[i])
#define RSMA_INFO_IS_DEL(r) ((r)->delFlag == 1)
#define RSMA_INFO_SET_DEL(r) ((r)->delFlag = 1)
#define RSMA_INFO_QTASK(r, i) ((r)->taskInfo[i])
#define RSMA_INFO_ITEM(r, i) (&(r)->items[i])
enum {
TASK_TRIGGER_STAT_INIT = 0,
@ -214,11 +219,11 @@ static FORCE_INLINE void tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat) {
int32_t smaPreClose(SSma *pSma);
// rsma
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer, int8_t rollback);
int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
// int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback);
void tdRSmaQTaskInfoGetFullPath(SVnode *pVnode, tb_uid_t suid, int8_t level, STfs *pTfs, char *outputName);

3
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -209,6 +209,7 @@ int32_t tsdbBegin(STsdb* pTsdb);
// int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
int32_t tsdbCacheCommit(STsdb* pTsdb);
int32_t tsdbCompact(STsdb* pTsdb, SCompactInfo* pInfo);
int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync);
// int32_t tsdbFinishCommit(STsdb* pTsdb);
// int32_t tsdbRollbackCommit(STsdb* pTsdb);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
@ -279,7 +280,7 @@ int32_t smaPrepareAsyncCommit(SSma* pSma);
int32_t smaCommit(SSma* pSma, SCommitInfo* pInfo);
int32_t smaFinishCommit(SSma* pSma);
int32_t smaPostCommit(SSma* pSma);
int32_t smaDoRetention(SSma* pSma, int64_t now);
int32_t smaRetention(SSma* pSma, int64_t now);
int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);

14
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -156,10 +156,10 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
nLoops = 0;
while (1) {
if (atomic_load_32(&pRSmaStat->nFetchAll) <= 0) {
smaDebug("vgId:%d, rsma commit:%d, fetch tasks are all finished", SMA_VID(pSma), isCommit);
smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are all finished", SMA_VID(pSma), isCommit);
break;
} else {
smaDebug("vgId:%d, rsma commit%d, fetch tasks are not all finished yet", SMA_VID(pSma), isCommit);
smaDebug("vgId:%d, rsma commit, type:%d, fetch tasks are not all finished yet", SMA_VID(pSma), isCommit);
}
TD_SMA_LOOPS_CHECK(nLoops, 1000);
}
@ -169,22 +169,24 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma, bool isCommit) {
* 1) This is high cost task and should not put in asyncPreCommit originally.
* 2) But, if put in asyncCommit, would trigger taskInfo cloning frequently.
*/
smaInfo("vgId:%d, rsma commit:%d, wait for all items to be consumed, TID:%p", SMA_VID(pSma), isCommit,
smaInfo("vgId:%d, rsma commit, type:%d, wait for all items to be consumed, TID:%p", SMA_VID(pSma), isCommit,
(void *)taosGetSelfPthreadId());
nLoops = 0;
while (atomic_load_64(&pRSmaStat->nBufItems) > 0) {
TD_SMA_LOOPS_CHECK(nLoops, 1000);
}
smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
if (!isCommit) goto _exit;
// code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
code = atomic_load_32(&pRSmaStat->execStat);
TSDB_CHECK_CODE(code, lino, _exit);
code = tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
TSDB_CHECK_CODE(code, lino, _exit);
smaInfo("vgId:%d, rsma commit, operator state committed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
smaInfo("vgId:%d, rsma commit, all items are consumed, TID:%p", SMA_VID(pSma), (void *)taosGetSelfPthreadId());
// all rsma results are written completely
STsdb *pTsdb = NULL;
if ((pTsdb = VND_RSMA1(pSma->pVnode))) {

9
source/dnode/vnode/src/sma/smaEnv.c
View File

@ -179,7 +179,7 @@ static void tRSmaInfoHashFreeNode(void *data) {
if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) {
taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
}
tdFreeRSmaInfo(pRSmaInfo->pSma, pRSmaInfo, true);
tdFreeRSmaInfo(pRSmaInfo->pSma, pRSmaInfo);
}
}
@ -209,6 +209,12 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS
pRSmaStat->pSma = (SSma *)pSma;
atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_INIT);
tsem_init(&pRSmaStat->notEmpty, 0, 0);
if (!(pRSmaStat->blocks = taosArrayInit(1, sizeof(SSDataBlock)))) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
SSDataBlock datablock = {.info.type = STREAM_CHECKPOINT};
taosArrayPush(pRSmaStat->blocks, &datablock);
// init smaMgmt
smaInit();
@ -290,6 +296,7 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
// step 5: free pStat
tsem_destroy(&(pStat->notEmpty));
taosArrayDestroy(pStat->blocks);
taosMemoryFreeClear(pStat);
}
}

390
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -15,12 +15,14 @@
#include "sma.h"
#include "tq.h"
#include "tstream.h"
#define RSMA_QTASKEXEC_SMOOTH_SIZE (100) // cnt
#define RSMA_SUBMIT_BATCH_SIZE (1024) // cnt
#define RSMA_FETCH_DELAY_MAX (120000) // ms
#define RSMA_FETCH_ACTIVE_MAX (1000) // ms
#define RSMA_FETCH_INTERVAL (5000) // ms
#define RSMA_TASK_FLAG "rsma"
#define RSMA_NEED_FETCH(r) (RSMA_INFO_ITEM((r), 0)->fetchLevel || RSMA_INFO_ITEM((r), 1)->fetchLevel)
@ -42,8 +44,8 @@ static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid);
static void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
static void tdFreeRSmaSubmitItems(SArray *pItems);
static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo);
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
int64_t suid);
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo,
int32_t execType, int8_t *streamFlushed);
static void tdRSmaFetchTrigger(void *param, void *tmrId);
static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables);
@ -72,41 +74,39 @@ static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t l
*
* @param pSma
* @param pInfo
* @param isDeepFree Only stop tmrId and free pTSchema for deep free
* @return void*
*/
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) {
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
if (pInfo) {
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = &pInfo->items[i];
if (isDeepFree && pItem->tmrId) {
if (pItem->tmrId) {
smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", SMA_VID(pSma), pItem->tmrId,
pInfo->suid, i + 1);
taosTmrStopA(&pItem->tmrId);
}
if (isDeepFree && pItem->pStreamState) {
if (pItem->pStreamState) {
streamStateClose(pItem->pStreamState, false);
}
if (isDeepFree && pInfo->taskInfo[i]) {
tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
if (pItem->pStreamTask) {
tFreeStreamTask(pItem->pStreamTask);
}
}
if (isDeepFree) {
taosMemoryFreeClear(pInfo->pTSchema);
taosArrayDestroy(pItem->pResList);
tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
}
if (isDeepFree) {
if (pInfo->queue) {
taosCloseQueue(pInfo->queue);
pInfo->queue = NULL;
}
if (pInfo->qall) {
taosFreeQall(pInfo->qall);
pInfo->qall = NULL;
}
taosMemoryFreeClear(pInfo->pTSchema);
if (pInfo->queue) {
taosCloseQueue(pInfo->queue);
pInfo->queue = NULL;
}
if (pInfo->qall) {
taosFreeQall(pInfo->qall);
pInfo->qall = NULL;
}
taosMemoryFree(pInfo);
@ -135,7 +135,9 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids,
return TSDB_CODE_FAILED;
}
if (!taosArrayGetSize(tbUids)) {
int32_t nTables = taosArrayGetSize(tbUids);
if (0 == nTables) {
smaDebug("vgId:%d, no need to update tbUidList for suid:%" PRIi64 " since Empty tbUids", SMA_VID(pSma), *suid);
return TSDB_CODE_SUCCESS;
}
@ -156,8 +158,9 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids,
terrstr());
return TSDB_CODE_FAILED;
}
smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 " uid:%" PRIi64 " level %d",
SMA_VID(pSma), pRSmaInfo->taskInfo[i], *suid, *(int64_t *)taosArrayGet(tbUids, 0), i);
smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p. suid:%" PRIi64 " uid:%" PRIi64
"nTables:%d level %d",
SMA_VID(pSma), pRSmaInfo->taskInfo[i], *suid, *(int64_t *)TARRAY_GET_ELEM(tbUids, 0), nTables, i);
}
}
@ -174,8 +177,8 @@ int32_t tdUpdateTbUidList(SSma *pSma, STbUidStore *pStore, bool isAdd) {
return TSDB_CODE_FAILED;
}
void *pIter = taosHashIterate(pStore->uidHash, NULL);
while (pIter) {
void *pIter = NULL;
while ((pIter = taosHashIterate(pStore->uidHash, pIter))) {
tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
SArray *pTbUids = *(SArray **)pIter;
@ -183,8 +186,6 @@ int32_t tdUpdateTbUidList(SSma *pSma, STbUidStore *pStore, bool isAdd) {
taosHashCancelIterate(pStore->uidHash, pIter);
return TSDB_CODE_FAILED;
}
pIter = taosHashIterate(pStore->uidHash, pIter);
}
return TSDB_CODE_SUCCESS;
}
@ -232,14 +233,37 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
return TSDB_CODE_SUCCESS;
}
static void tdRSmaTaskInit(SStreamMeta *pMeta, SRSmaInfoItem *pItem, SStreamTaskId *pId) {
STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
taosRLockLatch(&pMeta->lock);
SStreamTask **ppTask = (SStreamTask **)taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (ppTask && *ppTask) {
pItem->submitReqVer = (*ppTask)->chkInfo.checkpointVer;
pItem->fetchResultVer = (*ppTask)->info.triggerParam;
}
taosRUnLockLatch(&pMeta->lock);
}
static void tdRSmaTaskRemove(SStreamMeta *pMeta, int64_t streamId, int32_t taskId) {
streamMetaUnregisterTask(pMeta, streamId, taskId);
taosWLockLatch(&pMeta->lock);
int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
if (streamMetaCommit(pMeta) < 0) {
// persist to disk
}
taosWUnLockLatch(&pMeta->lock);
smaDebug("vgId:%d, rsma task:%" PRIi64 ",%d dropped, remain tasks:%d", pMeta->vgId, streamId, taskId, numOfTasks);
}
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
int8_t idx) {
if ((param->qmsgLen > 0) && param->qmsg[idx]) {
SRetention *pRetention = SMA_RETENTION(pSma);
STsdbCfg *pTsdbCfg = SMA_TSDB_CFG(pSma);
SVnode *pVnode = pSma->pVnode;
char taskInfDir[TSDB_FILENAME_LEN] = {0};
void *pStreamState = NULL;
SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
SRetention *pRetention = SMA_RETENTION(pSma);
STsdbCfg *pTsdbCfg = SMA_TSDB_CFG(pSma);
SVnode *pVnode = pSma->pVnode;
char taskInfDir[TSDB_FILENAME_LEN] = {0};
void *pStreamState = NULL;
// set the backend of stream state
tdRSmaQTaskInfoGetFullPath(pVnode, pRSmaInfo->suid, idx + 1, pVnode->pTfs, taskInfDir);
@ -254,25 +278,46 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
taosMemoryFree(s);
}
SStreamTask task = {.id.taskId = 0, .id.streamId = 0}; // TODO: assign value
task.pMeta = pVnode->pTq->pStreamMeta;
pStreamState = streamStateOpen(taskInfDir, &task, true, -1, -1);
SStreamTask *pStreamTask = taosMemoryCalloc(1, sizeof(*pStreamTask));
if (!pStreamTask) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
pItem->pStreamTask = pStreamTask;
pStreamTask->id.taskId = 0;
pStreamTask->id.streamId = pRSmaInfo->suid + idx;
pStreamTask->chkInfo.startTs = taosGetTimestampMs();
pStreamTask->pMeta = pVnode->pTq->pStreamMeta;
pStreamTask->exec.qmsg = taosMemoryMalloc(strlen(RSMA_TASK_FLAG) + 1);
sprintf(pStreamTask->exec.qmsg, "%s", RSMA_TASK_FLAG);
pStreamTask->chkInfo.checkpointId = streamMetaGetLatestCheckpointId(pStreamTask->pMeta);
tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id);
pStreamState = streamStateOpen(taskInfDir, pStreamTask, true, -1, -1);
if (!pStreamState) {
terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN;
return TSDB_CODE_FAILED;
}
pItem->pStreamState = pStreamState;
SReadHandle handle = {.vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState};
tdRSmaTaskRemove(pStreamTask->pMeta, pStreamTask->id.streamId, pStreamTask->id.taskId);
SReadHandle handle = {.vnode = pVnode, .initTqReader = 1, .skipRollup = 1, .pStateBackend = pStreamState};
initStorageAPI(&handle.api);
pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode), 0);
if (!pRSmaInfo->taskInfo[idx]) {
terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
return TSDB_CODE_FAILED;
}
SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
pItem->triggerStat = TASK_TRIGGER_STAT_ACTIVE; // fetch the data when reboot
pItem->pStreamState = pStreamState;
if (!(pItem->pResList = taosArrayInit(1, POINTER_BYTES))) {
return TSDB_CODE_FAILED;
}
if (pItem->fetchResultVer < pItem->submitReqVer) {
// fetch the data when reboot
pItem->triggerStat = TASK_TRIGGER_STAT_ACTIVE;
}
if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
int64_t msInterval =
convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
@ -291,10 +336,11 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
smaInfo("vgId:%d, item:%p table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
smaInfo("vgId:%d, open rsma task:%p table:%" PRIi64 " level:%" PRIi8 ", checkpointId:%" PRIi64
", submitReqVer:%" PRIi64 ", fetchResultVer:%" PRIi64 ", maxdelay:%" PRIi64 " watermark:%" PRIi64
", finally maxdelay:%" PRIi32,
TD_VID(pVnode), pItem, pRSmaInfo->suid, (int8_t)(idx + 1), param->maxdelay[idx], param->watermark[idx],
pItem->maxDelay);
TD_VID(pVnode), pItem->pStreamTask, pRSmaInfo->suid, (int8_t)(idx + 1), pStreamTask->chkInfo.checkpointId,
pItem->submitReqVer, pItem->fetchResultVer, param->maxdelay[idx], param->watermark[idx], pItem->maxDelay);
}
return TSDB_CODE_SUCCESS;
}
@ -362,7 +408,7 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
return TSDB_CODE_SUCCESS;
_err:
tdFreeRSmaInfo(pSma, pRSmaInfo, true);
tdFreeRSmaInfo(pSma, pRSmaInfo);
return TSDB_CODE_FAILED;
}
@ -498,11 +544,10 @@ static void tdUidStoreDestory(STbUidStore *pStore) {
if (pStore->uidHash) {
if (pStore->tbUids) {
// When pStore->tbUids not NULL, the pStore->uidHash has k/v; otherwise pStore->uidHash only has keys.
void *pIter = taosHashIterate(pStore->uidHash, NULL);
while (pIter) {
void *pIter = NULL;
while ((pIter = taosHashIterate(pStore->uidHash, pIter))) {
SArray *arr = *(SArray **)pIter;
taosArrayDestroy(arr);
pIter = taosHashIterate(pStore->uidHash, pIter);
}
}
taosHashCleanup(pStore->uidHash);
@ -562,7 +607,7 @@ static int32_t tdFetchSubmitReqSuids(SSubmitReq2 *pMsg, STbUidStore *pStore) {
* @param now
* @return int32_t
*/
int32_t smaDoRetention(SSma *pSma, int64_t now) {
int32_t smaRetention(SSma *pSma, int64_t now) {
int32_t code = TSDB_CODE_SUCCESS;
if (!VND_IS_RSMA(pSma->pVnode)) {
return code;
@ -570,8 +615,8 @@ int32_t smaDoRetention(SSma *pSma, int64_t now) {
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pSma->pRSmaTsdb[i]) {
// code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
// if (code) goto _end;
code = tsdbRetention(pSma->pRSmaTsdb[i], now, pSma->pVnode->config.sttTrigger == 1);
if (code) goto _end;
}
}
@ -579,17 +624,14 @@ _end:
return code;
}
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
int64_t suid) {
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo,
int32_t execType, int8_t *streamFlushed) {
int32_t code = 0;
int32_t lino = 0;
SSDataBlock *output = NULL;
SArray *pResList = taosArrayInit(1, POINTER_BYTES);
if (pResList == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
SArray *pResList = pItem->pResList;
STSchema *pTSchema = pInfo->pTSchema;
int64_t suid = pInfo->suid;
while (1) {
uint64_t ts;
@ -604,32 +646,46 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
if (taosArrayGetSize(pResList) == 0) {
break;
}
#if 0
char flag[10] = {0};
snprintf(flag, 10, "level %" PRIi8, pItem->level);
blockDebugShowDataBlocks(pResList, flag);
#endif
for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
output = taosArrayGetP(pResList, i);
if (output->info.type == STREAM_CHECKPOINT) {
if (streamFlushed) *streamFlushed = 1;
continue;
}
smaDebug("vgId:%d, result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64, SMA_VID(pSma),
output->info.id.uid, output->info.id.groupId, output->info.rows);
STsdb *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]);
SSubmitReq2 *pReq = NULL;
// TODO: the schema update should be handled later(TD-17965)
if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, output->info.id.groupId, SMA_VID(pSma), suid) < 0) {
code = terrno ? terrno : TSDB_CODE_RSMA_RESULT;
TSDB_CHECK_CODE(code, lino, _exit);
}
// reset the output version to handle reboot
if (STREAM_GET_ALL == execType && output->info.version == 0) {
// the submitReqVer keeps unchanged since tdExecuteRSmaImpl and tdRSmaFetchAllResult are executed synchronously
output->info.version = pItem->submitReqVer;
}
if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
code = terrno ? terrno : TSDB_CODE_RSMA_RESULT;
if (terrno == TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE) {
// TODO: reconfigure SSubmitReq2
} else {
if (terrno == 0) terrno = TSDB_CODE_RSMA_RESULT;
code = terrno;
}
tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFree(pReq);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (STREAM_GET_ALL == execType) {
atomic_store_64(&pItem->fetchResultVer, output->info.version);
}
smaDebug("vgId:%d, process submit req for rsma suid:%" PRIu64 ",uid:%" PRIu64 ", level %" PRIi8 " ver %" PRIi64,
SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version);
@ -648,7 +704,6 @@ _exit:
} else {
smaDebug("vgId:%d, %s succeed, suid:%" PRIi64 ", level:%" PRIi8, SMA_VID(pSma), __func__, suid, pItem->level);
}
taosArrayDestroy(pResList);
qCleanExecTaskBlockBuf(taskInfo);
return code;
}
@ -743,8 +798,9 @@ static int32_t tdRsmaPrintSubmitReq(SSma *pSma, SSubmitReq *pReq) {
*/
static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int32_t inputType, SRSmaInfo *pInfo,
ERsmaExecType type, int8_t level) {
int32_t idx = level - 1;
void *qTaskInfo = RSMA_INFO_QTASK(pInfo, idx);
int32_t idx = level - 1;
void *qTaskInfo = RSMA_INFO_QTASK(pInfo, idx);
SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, idx);
if (!qTaskInfo) {
smaDebug("vgId:%d, no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level,
@ -772,10 +828,14 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize,
return TSDB_CODE_FAILED;
}
SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, idx);
tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo->pTSchema, pInfo->suid);
if (STREAM_INPUT__MERGED_SUBMIT == inputType) {
SPackedData *packData = POINTER_SHIFT(pMsg, sizeof(SPackedData) * (msgSize - 1));
atomic_store_64(&pItem->submitReqVer, packData->ver);
}
return TSDB_CODE_SUCCESS;
terrno = tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo, STREAM_NORMAL, NULL);
return terrno ? TSDB_CODE_FAILED : TDB_CODE_SUCCESS;
}
/**
@ -881,7 +941,12 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg,
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
if (!pEnv) {
// only applicable when rsma env exists
return TSDB_CODE_SUCCESS;
return TDB_CODE_SUCCESS;
}
if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) {
smaError("vgId:%d, failed to process rsma submit since invalid exec code: %s", SMA_VID(pSma), terrstr());
goto _err;
}
STbUidStore uidStore = {0};
@ -913,7 +978,7 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg,
return TSDB_CODE_SUCCESS;
_err:
tdUidStoreDestory(&uidStore);
return TSDB_CODE_FAILED;
return terrno;
}
/**
@ -988,7 +1053,7 @@ static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables) {
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
#if 0
// reload all ctbUids for suid
uidStore.suid = suid;
if (vnodeGetCtbIdList(pVnode, suid, uidStore.tbUids) < 0) {
@ -1002,7 +1067,7 @@ static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables) {
}
taosArrayClear(uidStore.tbUids);
#endif
smaDebug("vgId:%d, rsma restore env success for %" PRIi64, TD_VID(pVnode), suid);
}
}
@ -1050,39 +1115,141 @@ _err:
return code;
}
#if 0
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
int32_t code = 0;
int32_t lino = 0;
int32_t nTaskInfo = 0;
SSma *pSma = pRSmaStat->pSma;
SVnode *pVnode = pSma->pVnode;
SRSmaFS fs = {0};
if (taosHashGetSize(pInfoHash) <= 0) {
return TSDB_CODE_SUCCESS;
}
void *infoHash = NULL;
while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
continue;
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pRSmaInfo, i);
if (pItem && pItem->pStreamState) {
if (streamStateCommit(pItem->pStreamState) < 0) {
code = TSDB_CODE_RSMA_STREAM_STATE_COMMIT;
TSDB_CHECK_CODE(code, lino, _exit);
// stream state: trigger checkpoint
do {
void *infoHash = NULL;
while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
continue;
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pRSmaInfo->taskInfo[i]) {
code = qSetSMAInput(pRSmaInfo->taskInfo[i], pRSmaStat->blocks, 1, STREAM_INPUT__CHECKPOINT);
if (code) {
taosHashCancelIterate(pInfoHash, infoHash);
TSDB_CHECK_CODE(code, lino, _exit);
}
pRSmaInfo->items[i].streamFlushed = 0;
++nTaskInfo;
}
smaDebug("vgId:%d, rsma persist, stream state commit success, table %" PRIi64 ", level %d", TD_VID(pVnode),
pRSmaInfo->suid, i + 1);
}
}
}
} while (0);
// stream state: wait checkpoint ready in async mode
do {
int32_t nStreamFlushed = 0;
int32_t nSleep = 0;
void *infoHash = NULL;
while (true) {
while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
continue;
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
if (pRSmaInfo->taskInfo[i] && (0 == pRSmaInfo->items[i].streamFlushed)) {
int8_t streamFlushed = 0;
code = tdRSmaExecAndSubmitResult(pSma, pRSmaInfo->taskInfo[i], &pRSmaInfo->items[i], pRSmaInfo,
STREAM_CHECKPOINT, &streamFlushed);
if (code) {
taosHashCancelIterate(pInfoHash, infoHash);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (streamFlushed) {
pRSmaInfo->items[i].streamFlushed = 1;
if (++nStreamFlushed >= nTaskInfo) {
smaInfo("vgId:%d, rsma commit, checkpoint ready, %d us consumed, received/total: %d/%d", TD_VID(pVnode),
nSleep * 10, nStreamFlushed, nTaskInfo);
taosHashCancelIterate(pInfoHash, infoHash);
goto _checkpoint;
}
}
}
}
}
taosUsleep(10);
++nSleep;
smaDebug("vgId:%d, rsma commit, wait for checkpoint ready, %d us elapsed, received/total: %d/%d", TD_VID(pVnode),
nSleep * 10, nStreamFlushed, nTaskInfo);
}
} while (0);
_checkpoint:
// stream state: build checkpoint in backend
do {
SStreamMeta *pMeta = NULL;
int64_t checkpointId = taosGetTimestampNs();
bool checkpointBuilt = false;
void *infoHash = NULL;
while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
continue;
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pRSmaInfo, i);
if (pItem && pItem->pStreamTask) {
SStreamTask *pTask = pItem->pStreamTask;
atomic_store_32(&pTask->pMeta->chkptNotReadyTasks, 1);
pTask->checkpointingId = checkpointId;
pTask->chkInfo.checkpointId = pTask->checkpointingId;
pTask->chkInfo.checkpointVer = pItem->submitReqVer;
pTask->info.triggerParam = pItem->fetchResultVer;
if (!checkpointBuilt) {
// the stream states share one checkpoint
code = streamTaskBuildCheckpoint(pTask);
if (code) {
taosHashCancelIterate(pInfoHash, infoHash);
TSDB_CHECK_CODE(code, lino, _exit);
}
pMeta = pTask->pMeta;
checkpointBuilt = true;
}
taosWLockLatch(&pMeta->lock);
if (streamMetaSaveTask(pMeta, pTask)) {
taosWUnLockLatch(&pMeta->lock);
code = terrno ? terrno : TSDB_CODE_OUT_OF_MEMORY;
taosHashCancelIterate(pInfoHash, infoHash);
TSDB_CHECK_CODE(code, lino, _exit);
}
taosWUnLockLatch(&pMeta->lock);
smaDebug("vgId:%d, rsma commit, succeed to commit task:%p, submitReqVer:%" PRIi64 ", fetchResultVer:%" PRIi64
", table:%" PRIi64 ", level:%d",
TD_VID(pVnode), pTask, pItem->submitReqVer, pItem->fetchResultVer, pRSmaInfo->suid, i + 1);
}
}
}
if (pMeta) {
taosWLockLatch(&pMeta->lock);
if (streamMetaCommit(pMeta)) {
taosWUnLockLatch(&pMeta->lock);
code = terrno ? terrno : TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _exit);
}
taosWUnLockLatch(&pMeta->lock);
}
if (checkpointBuilt) {
smaInfo("vgId:%d, rsma commit, succeed to commit checkpoint:%" PRIi64, TD_VID(pVnode), checkpointId);
}
} while (0);
_exit:
if (code) {
smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
@ -1091,7 +1258,7 @@ _exit:
terrno = code;
return code;
}
#endif
/**
* @brief trigger to get rsma result in async mode
*
@ -1165,14 +1332,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active",
SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
// async process
pItem->fetchLevel = pItem->level;
#if 0
// debugging codes
SRSmaInfo *qInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaInfo->suid);
SRSmaInfoItem *qItem = RSMA_INFO_ITEM(qInfo, pItem->level - 1);
make sure(qItem->level == pItem->level);
make sure(qItem->fetchLevel == pItem->fetchLevel);
#endif
atomic_store_8(&pItem->fetchLevel, 1);
if (atomic_load_8(&pRSmaInfo->assigned) == 0) {
tsem_post(&(pStat->notEmpty));
}
@ -1217,15 +1378,15 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
for (int8_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, i - 1);
if (pItem->fetchLevel) {
pItem->fetchLevel = 0;
if (1 == atomic_val_compare_exchange_8(&pItem->fetchLevel, 1, 0)) {
qTaskInfo_t taskInfo = RSMA_INFO_QTASK(pInfo, i - 1);
if (!taskInfo) {
continue;
}
if ((++pItem->nScanned * pItem->maxDelay) > RSMA_FETCH_DELAY_MAX) {
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch executed",
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi32 " maxDelay:%d, fetch executed",
SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
} else {
int64_t curMs = taosGetTimestampMs();
@ -1245,14 +1406,15 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
if ((terrno = qSetSMAInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
goto _err;
}
if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid) < 0) {
if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo, STREAM_GET_ALL, NULL) < 0) {
atomic_store_32(&SMA_RSMA_STAT(pSma)->execStat, terrno);
goto _err;
}
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch finished",
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi32 " maxDelay:%d, fetch finished",
SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
} else {
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16
smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi32
" maxDelay:%d, fetch not executed as fetch level is %" PRIi8,
SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay, pItem->fetchLevel);
}
@ -1275,6 +1437,7 @@ static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SA
.msgStr = POINTER_SHIFT(msg, sizeof(int32_t) + sizeof(int64_t))};
if (!taosArrayPush(pSubmitArr, &packData)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tdFreeRSmaSubmitItems(pSubmitArr);
goto _err;
}
@ -1294,6 +1457,7 @@ static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SA
}
return TSDB_CODE_SUCCESS;
_err:
atomic_store_32(&SMA_RSMA_STAT(pSma)->execStat, terrno);
smaError("vgId:%d, batch exec for suid:%" PRIi64 " execType:%d size:%d failed since %s", SMA_VID(pSma), pInfo->suid,
type, (int32_t)taosArrayGetSize(pSubmitArr), terrstr());
tdFreeRSmaSubmitItems(pSubmitArr);
@ -1369,15 +1533,11 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
if (ASSERTS(oldVal >= 0, "oldVal of nFetchAll: %d < 0", oldVal)) {
code = TSDB_CODE_APP_ERROR;
taosHashCancelIterate(infoHash, pIter);
TSDB_CHECK_CODE(code, lino, _exit);
}
int8_t curStat = atomic_load_8(RSMA_COMMIT_STAT(pRSmaStat));
if (curStat == 1) {
smaDebug("vgId:%d, fetch all not exec as commit stat is %" PRIi8, SMA_VID(pSma), curStat);
} else {
tdRSmaFetchAllResult(pSma, pInfo);
}
tdRSmaFetchAllResult(pSma, pInfo);
if (0 == atomic_sub_fetch_32(&pRSmaStat->nFetchAll, 1)) {
atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);

2
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -12,11 +12,11 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "cos.h"
#include "tsdb.h"
#include "tsdbDataFileRW.h"
#include "tsdbReadUtil.h"
#include "vnd.h"
#include "vndCos.h"
#define ROCKS_BATCH_SIZE (4096)

2
source/dnode/vnode/src/tsdb/tsdbFS2.c
View File

@ -14,9 +14,9 @@
*/
#include "tsdbFS2.h"
#include "cos.h"
#include "tsdbUpgrade.h"
#include "vnd.h"
#include "vndCos.h"
#define BLOCK_COMMIT_FACTOR 3

2
source/dnode/vnode/src/tsdb/tsdbOpen.c
View File

@ -89,6 +89,8 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
return 0;
_err:
tsdbCloseFS(&pTsdb->pFS);
taosThreadMutexDestroy(&pTsdb->mutex);
taosMemoryFree(pTsdb);
return -1;
}

14
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -48,9 +48,9 @@ static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScan
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
STsdbReader* pReader);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SCostSummary* pCost);
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idstr,
int8_t* pLevel);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SCostSummary* pCost);
static STsdb* getTsdbByRetentions(SVnode* pVnode, SQueryTableDataCond* pCond, SRetention* retentions, const char* idstr,
int8_t* pLevel);
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
static int32_t doBuildDataBlock(STsdbReader* pReader);
@ -384,7 +384,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, void
initReaderStatus(&pReader->status);
pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->pTsdb = getTsdbByRetentions(pVnode, pCond, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->info.suid = pCond->suid;
pReader->info.order = pCond->order;
@ -3152,9 +3152,9 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
}
}
static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* retentions, const char* idStr,
static STsdb* getTsdbByRetentions(SVnode* pVnode, SQueryTableDataCond* pCond, SRetention* retentions, const char* idStr,
int8_t* pLevel) {
if (VND_IS_RSMA(pVnode)) {
if (VND_IS_RSMA(pVnode) && !pCond->skipRollup) {
int8_t level = 0;
int8_t precision = pVnode->config.tsdbCfg.precision;
int64_t now = taosGetTimestamp(precision);
@ -3170,7 +3170,7 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* ret
}
break;
}
if ((now - pRetention->keep) <= (winSKey + offset)) {
if ((now - pRetention->keep) <= (pCond->twindows.skey + offset)) {
break;
}
++level;

2
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "cos.h"
#include "tsdb.h"
#include "vndCos.h"
static int32_t tsdbOpenFileImpl(STsdbFD *pFD) {
int32_t code = 0;

8
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -15,7 +15,8 @@
#include "tsdb.h"
#include "tsdbFS2.h"
#include "vndCos.h"
#include "cos.h"
#include "vnd.h"
typedef struct {
STsdb *tsdb;
@ -387,6 +388,8 @@ _exit:
return code;
}
static void tsdbFreeRtnArg(void *arg) { taosMemoryFree(arg); }
static int32_t tsdbDoRetentionSync(void *arg) {
int32_t code = 0;
int32_t lino = 0;
@ -409,6 +412,7 @@ _exit:
TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
}
tsem_post(&((SRtnArg *)arg)->tsdb->pVnode->canCommit);
tsdbFreeRtnArg(arg);
return code;
}
@ -438,7 +442,7 @@ _exit:
return code;
}
static void tsdbFreeRtnArg(void *arg) { taosMemoryFree(arg); }
int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync) {
int32_t code = 0;

4
source/dnode/vnode/src/tsdb/tsdbWrite.c
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "cos.h"
#include "tsdb.h"
#include "vndCos.h"
/**
* @brief max key by precision
@ -39,7 +39,7 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq2 *pMsg, SSubmitRsp2
arrSize = taosArrayGetSize(pMsg->aSubmitTbData);
// scan and convert
if (tsdbScanAndConvertSubmitMsg(pTsdb, pMsg) < 0) {
if ((terrno = tsdbScanAndConvertSubmitMsg(pTsdb, pMsg)) < 0) {
if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
tsdbError("vgId:%d, failed to insert data since %s", TD_VID(pTsdb->pVnode), tstrerror(terrno));
}

2
source/dnode/vnode/src/vnd/vnodeModule.c
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "cos.h"
#include "vnd.h"
#include "vndCos.h"
typedef struct SVnodeTask SVnodeTask;
struct SVnodeTask {

2
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -13,10 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "cos.h"
#include "sync.h"
#include "tsdb.h"
#include "vnd.h"
#include "vndCos.h"
int32_t vnodeGetPrimaryDir(const char *relPath, int32_t diskPrimary, STfs *pTfs, char *buf, size_t bufLen) {
if (pTfs) {

10
source/dnode/vnode/src/vnd/vnodeRetention.c
View File

@ -15,8 +15,12 @@
#include "vnd.h"
extern int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync);
int32_t vnodeDoRetention(SVnode *pVnode, int64_t now) {
return tsdbRetention(pVnode->pTsdb, now, pVnode->config.sttTrigger == 1);
int32_t code = TSDB_CODE_SUCCESS;
code = tsdbRetention(pVnode->pTsdb, now, pVnode->config.sttTrigger == 1);
if (TSDB_CODE_SUCCESS == code) code = smaRetention(pVnode->pSma, now);
return code;
}

4
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -18,7 +18,7 @@
#include "tmsg.h"
#include "tstrbuild.h"
#include "vnd.h"
#include "vndCos.h"
#include "cos.h"
#include "vnode.h"
#include "vnodeInt.h"
@ -1669,7 +1669,7 @@ _exit:
atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1);
if (code == 0) {
atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1);
tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len, STREAM_INPUT__DATA_SUBMIT);
code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len, STREAM_INPUT__DATA_SUBMIT);
}
// clear

2
source/libs/executor/inc/executil.h
View File

@ -180,7 +180,7 @@ void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode);
SColumn extractColumnFromColumnNode(SColumnNode* pColNode);
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode, const SReadHandle* readHandle);
void cleanupQueryTableDataCond(SQueryTableDataCond* pCond);
int32_t convertFillType(int32_t mode);

3
source/libs/executor/src/executil.c
View File

@ -1713,7 +1713,7 @@ SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
return c;
}
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) {
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode, const SReadHandle* readHandle) {
pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
@ -1732,6 +1732,7 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
pCond->type = TIMEWINDOW_RANGE_CONTAINED;
pCond->startVersion = -1;
pCond->endVersion = -1;
pCond->skipRollup = readHandle->skipRollup;
int32_t j = 0;
for (int32_t i = 0; i < pCond->numOfCols; ++i) {

10
source/libs/executor/src/executor.c
View File

@ -69,12 +69,14 @@ static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOf
} else if (type == STREAM_INPUT__DATA_BLOCK) {
for (int32_t i = 0; i < numOfBlocks; ++i) {
SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
SPackedData tmp = {
.pDataBlock = pDataBlock,
};
SPackedData tmp = {.pDataBlock = pDataBlock};
taosArrayPush(pInfo->pBlockLists, &tmp);
}
pInfo->blockType = STREAM_INPUT__DATA_BLOCK;
} else if (type == STREAM_INPUT__CHECKPOINT) {
SPackedData tmp = {.pDataBlock = input};
taosArrayPush(pInfo->pBlockLists, &tmp);
pInfo->blockType = STREAM_INPUT__CHECKPOINT;
}
return TSDB_CODE_SUCCESS;
@ -633,7 +635,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bo
blockIndex += 1;
current += p->info.rows;
ASSERT(p->info.rows > 0);
ASSERT(p->info.rows > 0 || p->info.type == STREAM_CHECKPOINT);
taosArrayPush(pResList, &p);
if (current >= rowsThreshold) {

4
source/libs/executor/src/scanoperator.c
View File

@ -1035,7 +1035,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
}
initLimitInfo(pScanNode->node.pLimit, pScanNode->node.pSlimit, &pInfo->base.limitInfo);
code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode);
code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
@ -3533,7 +3533,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
goto _error;
}
code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode);
code = initQueryTableDataCond(&pInfo->base.cond, pTableScanNode, readHandle);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(pInfo->base.matchInfo.pList);
goto _error;

13
source/libs/stream/inc/streamInt.h
View File

@ -106,7 +106,6 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq);
int32_t streamSaveAllTaskStatus(SStreamMeta* pMeta, int64_t checkpointId);
int32_t streamTaskBuildCheckpoint(SStreamTask* pTask);
int32_t streamSendCheckMsg(SStreamTask* pTask, const SStreamTaskCheckReq* pReq, int32_t nodeId, SEpSet* pEpSet);
int32_t streamAddCheckpointReadyMsg(SStreamTask* pTask, int32_t srcTaskId, int32_t index, int64_t checkpointId);
@ -140,6 +139,18 @@ void* streamQueueNextItem(SStreamQueue* pQueue);
void streamFreeQitem(SStreamQueueItem* data);
int32_t streamQueueGetItemSize(const SStreamQueue* pQueue);
typedef enum UPLOAD_TYPE{
UPLOAD_DISABLE = -1,
UPLOAD_S3 = 0,
UPLOAD_RSYNC = 1,
} UPLOAD_TYPE;
UPLOAD_TYPE getUploadType();
int uploadCheckpoint(char* id, char* path);
int downloadCheckpoint(char* id, char* path);
int deleteCheckpoint(char* id);
int deleteCheckpointFile(char* id, char* name);
int32_t onNormalTaskReady(SStreamTask* pTask);
int32_t onScanhistoryTaskReady(SStreamTask* pTask);

90
source/libs/stream/src/streamCheckpoint.c
View File

@ -14,6 +14,8 @@
*/
#include "streamInt.h"
#include "rsync.h"
#include "cos.h"
int32_t tEncodeStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq) {
if (tStartEncode(pEncoder) < 0) return -1;
@ -372,3 +374,91 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
return code;
}
static int uploadCheckpointToS3(char* id, char* path){
TdDirPtr pDir = taosOpenDir(path);
if (pDir == NULL) return -1;
TdDirEntryPtr de = NULL;
while ((de = taosReadDir(pDir)) != NULL) {
char* name = taosGetDirEntryName(de);
if (strcmp(name, ".") == 0 || strcmp(name, "..") == 0 ||
taosDirEntryIsDir(de)) continue;
char filename[PATH_MAX] = {0};
if(path[strlen(path) - 1] == TD_DIRSEP_CHAR){
snprintf(filename, sizeof(filename), "%s%s", path, name);
}else{
snprintf(filename, sizeof(filename), "%s%s%s", path, TD_DIRSEP, name);
}
char object[PATH_MAX] = {0};
snprintf(object, sizeof(object), "%s%s%s", id, TD_DIRSEP, name);
if(s3PutObjectFromFile2(filename, object) != 0){
taosCloseDir(&pDir);
return -1;
}
stDebug("[s3] upload checkpoint:%s", filename);
}
taosCloseDir(&pDir);
return 0;
}
UPLOAD_TYPE getUploadType(){
if(strlen(tsSnodeAddress) != 0){
return UPLOAD_RSYNC;
}else if(tsS3StreamEnabled){
return UPLOAD_S3;
}else{
return UPLOAD_DISABLE;
}
}
int uploadCheckpoint(char* id, char* path){
if(id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX){
stError("uploadCheckpoint parameters invalid");
return -1;
}
if(strlen(tsSnodeAddress) != 0){
return uploadRsync(id, path);
}else if(tsS3StreamEnabled){
return uploadCheckpointToS3(id, path);
}
return 0;
}
int downloadCheckpoint(char* id, char* path){
if(id == NULL || path == NULL || strlen(id) == 0 || strlen(path) == 0 || strlen(path) >= PATH_MAX){
stError("downloadCheckpoint parameters invalid");
return -1;
}
if(strlen(tsSnodeAddress) != 0){
return downloadRsync(id, path);
}else if(tsS3StreamEnabled){
return s3GetObjectsByPrefix(id, path);
}
return 0;
}
int deleteCheckpoint(char* id){
if(id == NULL || strlen(id) == 0){
stError("deleteCheckpoint parameters invalid");
return -1;
}
if(strlen(tsSnodeAddress) != 0){
return deleteRsync(id);
}else if(tsS3StreamEnabled){
s3DeleteObjectsByPrefix(id);
}
return 0;
}
int deleteCheckpointFile(char* id, char* name){
char object[128] = {0};
snprintf(object, sizeof(object), "%s/%s", id, name);
char *tmp = object;
s3DeleteObjects((const char**)&tmp, 1);
return 0;
}

7
source/libs/stream/src/streamMeta.c
View File

@ -28,7 +28,6 @@ int32_t streamBackendId = 0;
int32_t streamBackendCfWrapperId = 0;
int32_t streamMetaId = 0;
static int64_t streamGetLatestCheckpointId(SStreamMeta* pMeta);
static void metaHbToMnode(void* param, void* tmrId);
static void streamMetaClear(SStreamMeta* pMeta);
static int32_t streamMetaBegin(SStreamMeta* pMeta);
@ -191,10 +190,10 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
pMeta->chkpSaved = taosArrayInit(4, sizeof(int64_t));
pMeta->chkpInUse = taosArrayInit(4, sizeof(int64_t));
pMeta->chkpCap = 8;
pMeta->chkpCap = 2;
taosInitRWLatch(&pMeta->chkpDirLock);
pMeta->chkpId = streamGetLatestCheckpointId(pMeta);
pMeta->chkpId = streamMetaGetLatestCheckpointId(pMeta);
pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId);
while (pMeta->streamBackend == NULL) {
taosMsleep(100);
@ -602,7 +601,7 @@ int32_t streamMetaCommit(SStreamMeta* pMeta) {
return 0;
}
int64_t streamGetLatestCheckpointId(SStreamMeta* pMeta) {
int64_t streamMetaGetLatestCheckpointId(SStreamMeta* pMeta) {
int64_t chkpId = 0;
TBC* pCur = NULL;

5
source/libs/stream/src/streamSnapshot.c
View File

@ -51,6 +51,7 @@ struct SStreamSnapHandle {
int8_t filetype;
SArray* pFileList;
int32_t currFileIdx;
int8_t delFlag; // 0 : not del, 1: del
};
struct SStreamSnapBlockHdr {
int8_t type;
@ -147,6 +148,7 @@ int32_t streamSnapHandleInit(SStreamSnapHandle* pHandle, char* path, int64_t chk
taosMemoryFree(tdir);
return code;
}
pHandle->delFlag = 1;
chkpId = 0;
}
@ -273,7 +275,7 @@ void streamSnapHandleDestroy(SStreamSnapHandle* handle) {
if (handle->checkpointId == 0) {
// del tmp dir
if (pFile && taosIsDir(pFile->path)) {
taosRemoveDir(pFile->path);
if (handle->delFlag) taosRemoveDir(pFile->path);
}
} else {
streamBackendDelInUseChkp(handle->handle, handle->checkpointId);
@ -422,6 +424,7 @@ int32_t streamSnapWriterOpen(void* pMeta, int64_t sver, int64_t ever, char* path
pHandle->pFileList = list;
pHandle->currFileIdx = 0;
pHandle->offset = 0;
pHandle->delFlag = 0;
*ppWriter = pWriter;
return 0;

3
source/libs/stream/src/streamState.c
View File

@ -221,7 +221,6 @@ SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t sz
}
pState->pTdbState->pOwner = pTask;
pState->checkPointId = 0;
return pState;
@ -274,7 +273,6 @@ int32_t streamStateCommit(SStreamState* pState) {
SStreamSnapshot* pShot = getSnapshot(pState->pFileState);
flushSnapshot(pState->pFileState, pShot, true);
}
pState->checkPointId++;
return 0;
#else
if (tdbCommit(pState->pTdbState->db, pState->pTdbState->txn) < 0) {
@ -288,7 +286,6 @@ int32_t streamStateCommit(SStreamState* pState) {
TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
return -1;
}
pState->checkPointId++;
return 0;
#endif
}

66
source/libs/stream/src/tstreamFileState.c
View File

@ -27,6 +27,9 @@
#define DEFAULT_MAX_STREAM_BUFFER_SIZE (128 * 1024 * 1024)
#define MIN_NUM_OF_ROW_BUFF 10240
#define TASK_KEY "streamFileState"
#define STREAM_STATE_INFO_NAME "StreamStateCheckPoint"
struct SStreamFileState {
SList* usedBuffs;
SList* freeBuffs;
@ -113,6 +116,15 @@ void* sessionCreateStateKey(SRowBuffPos* pPos, int64_t num) {
return pStateKey;
}
static void streamFileStateDecode(TSKEY* pKey, void* pBuff, int32_t len) { pBuff = taosDecodeFixedI64(pBuff, pKey); }
static void streamFileStateEncode(TSKEY* pKey, void** pVal, int32_t* pLen) {
*pLen = sizeof(TSKEY);
(*pVal) = taosMemoryCalloc(1, *pLen);
void* buff = *pVal;
taosEncodeFixedI64(&buff, *pKey);
}
SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize,
GetTsFun fp, void* pFile, TSKEY delMark, const char* taskId, int64_t checkpointId,
int8_t type) {
@ -181,6 +193,15 @@ SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_
recoverSesssion(pFileState, checkpointId);
}
void* valBuf = NULL;
int32_t len = 0;
int32_t code = streamDefaultGet_rocksdb(pFileState->pFileStore, STREAM_STATE_INFO_NAME, &valBuf, &len);
if (code == TSDB_CODE_SUCCESS) {
ASSERT(len == sizeof(TSKEY));
streamFileStateDecode(&pFileState->flushMark, valBuf, len);
qDebug("===stream===flushMark read:%" PRId64, pFileState->flushMark);
}
taosMemoryFreeClear(valBuf);
return pFileState;
_error:
@ -506,15 +527,6 @@ SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {
return pFileState->usedBuffs;
}
void streamFileStateDecode(TSKEY* pKey, void* pBuff, int32_t len) { pBuff = taosDecodeFixedI64(pBuff, pKey); }
void streamFileStateEncode(TSKEY* pKey, void** pVal, int32_t* pLen) {
*pLen = sizeof(TSKEY);
(*pVal) = taosMemoryCalloc(1, *pLen);
void* buff = *pVal;
taosEncodeFixedI64(&buff, *pKey);
}
int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {
int32_t code = TSDB_CODE_SUCCESS;
SListIter iter = {0};
@ -538,6 +550,7 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
continue;
}
pPos->beFlushed = true;
pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));
qDebug("===stream===flushed start:%" PRId64, pFileState->getTs(pPos->pKey));
if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {
@ -565,24 +578,12 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
pFileState->id, numOfElems, BATCH_LIMIT, elapsed);
if (flushState) {
const char* taskKey = "streamFileState";
{
char keyBuf[128] = {0};
void* valBuf = NULL;
int32_t len = 0;
sprintf(keyBuf, "%s:%" PRId64 "", taskKey, ((SStreamState*)pFileState->pFileStore)->checkPointId);
streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);
streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len, 0);
taosMemoryFree(valBuf);
}
{
char keyBuf[128] = {0};
char valBuf[64] = {0};
int32_t len = 0;
memcpy(keyBuf, taskKey, strlen(taskKey));
len = sprintf(valBuf, "%" PRId64 "", ((SStreamState*)pFileState->pFileStore)->checkPointId);
code = streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len, 0);
}
void* valBuf = NULL;
int32_t len = 0;
streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);
qDebug("===stream===flushMark write:%" PRId64, pFileState->flushMark);
streamStatePutBatch(pFileState->pFileStore, "default", batch, STREAM_STATE_INFO_NAME, valBuf, len, 0);
taosMemoryFree(valBuf);
streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
}
@ -591,26 +592,23 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
}
int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {
const char* taskKey = "streamFileState";
char keyBuf[128] = {0};
sprintf(keyBuf, "%s:%" PRId64 "", taskKey, checkpointId);
sprintf(keyBuf, "%s:%" PRId64 "", TASK_KEY, checkpointId);
return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);
}
int32_t getSnapshotIdList(SStreamFileState* pFileState, SArray* list) {
const char* taskKey = "streamFileState";
return streamDefaultIterGet_rocksdb(pFileState->pFileStore, taskKey, NULL, list);
return streamDefaultIterGet_rocksdb(pFileState->pFileStore, TASK_KEY, NULL, list);
}
int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
int32_t code = TSDB_CODE_SUCCESS;
const char* taskKey = "streamFileState";
int64_t maxCheckPointId = 0;
{
char buf[128] = {0};
void* val = NULL;
int32_t len = 0;
memcpy(buf, taskKey, strlen(taskKey));
memcpy(buf, TASK_KEY, strlen(TASK_KEY));
code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
if (code != 0 || len == 0 || val == NULL) {
return TSDB_CODE_FAILED;
@ -624,7 +622,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
char buf[128] = {0};
void* val = 0;
int32_t len = 0;
sprintf(buf, "%s:%" PRId64 "", taskKey, i);
sprintf(buf, "%s:%" PRId64 "", TASK_KEY, i);
code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
if (code != 0) {
return TSDB_CODE_FAILED;

11
source/libs/stream/test/CMakeLists.txt
View File

@ -18,6 +18,17 @@ TARGET_INCLUDE_DIRECTORIES(
PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
)
ADD_EXECUTABLE(checkpointTest checkpointTest.cpp)
TARGET_LINK_LIBRARIES(
checkpointTest
PUBLIC os common gtest stream executor qcom index transport util
)
TARGET_INCLUDE_DIRECTORIES(
checkpointTest
PRIVATE "${TD_SOURCE_DIR}/source/libs/stream/inc"
)
add_test(
NAME streamUpdateTest
COMMAND streamUpdateTest

68
source/libs/stream/test/checkpointTest.cpp Normal file
View File

@ -0,0 +1,68 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <taoserror.h>
#include <tglobal.h>
#include <iostream>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
#pragma GCC diagnostic ignored "-Wunused-variable"
#pragma GCC diagnostic ignored "-Wsign-compare"
#include "rsync.h"
#include "streamInt.h"
#include "cos.h"
int main(int argc, char **argv) {
testing::InitGoogleTest(&argc, argv);
if (taosInitCfg("/etc/taos/", NULL, NULL, NULL, NULL, 0) != 0) {
printf("error");
}
if (s3Init() < 0) {
return -1;
}
strcpy(tsSnodeAddress, "127.0.0.1");
return RUN_ALL_TESTS();
}
TEST(testCase, checkpointUpload_Test) {
stopRsync();
startRsync();
taosSsleep(5);
char* id = "2013892036";
uploadCheckpoint(id, "/root/offset/");
}
TEST(testCase, checkpointDownload_Test) {
char* id = "2013892036";
downloadCheckpoint(id, "/root/offset/download/");
}
TEST(testCase, checkpointDelete_Test) {
char* id = "2013892036";
deleteCheckpoint(id);
}
TEST(testCase, checkpointDeleteFile_Test) {
char* id = "2013892036";
deleteCheckpointFile(id, "offset-ver0");
}

2
tests/parallel_test/cases.task
View File

@ -229,6 +229,7 @@ e
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/ttlChangeOnWrite.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/compress_tsz1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/compress_tsz2.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/view/non_marterial_view/test_view.py
,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
,,n,system-test,python3 ./test.py -f 0-others/tag_index_basic.py
,,n,system-test,python3 ./test.py -f 0-others/udfpy_main.py
@ -1181,6 +1182,7 @@ e
,,y,script,./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
,,y,script,./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
,,y,script,./test.sh -f tsim/sync/vnodesnapshot-rsma-test.sim
,,n,script,./test.sh -f tsim/valgrind/checkError1.sim
,,n,script,./test.sh -f tsim/valgrind/checkError2.sim
,,n,script,./test.sh -f tsim/valgrind/checkError3.sim

4
tests/pytest/util/sql.py
View File

@ -88,7 +88,7 @@ class TDSql:
expectErrNotOccured = False
self.errno = e.errno
error_info = repr(e)
self.error_info = error_info[error_info.index('(')+1:-1].split(",")[0].replace("'","")
self.error_info = ','.join(error_info[error_info.index('(')+1:-1].split(",")[:-1]).replace("'","")
# self.error_info = (','.join(error_info.split(",")[:-1]).split("(",1)[1:][0]).replace("'","")
if expectErrNotOccured:
tdLog.exit("%s(%d) failed: sql:%s, expect error not occured" % (caller.filename, caller.lineno, sql))
@ -106,7 +106,7 @@ class TDSql:
tdLog.info("sql:%s, expect error occured" % (sql))
if expectErrInfo != None:
if expectErrInfo == self.error_info:
if expectErrInfo == self.error_info or expectErrInfo in self.error_info:
tdLog.info("sql:%s, expected expectErrInfo %s occured" % (sql, expectErrInfo))
else:
tdLog.exit("%s(%d) failed: sql:%s, expectErrInfo %s occured, but not expected errno %s" % (caller.filename, caller.lineno, sql, self.error_info, expectErrInfo))

2
tests/script/tsim/sma/rsmaPersistenceRecovery.sim
View File

@ -5,7 +5,7 @@ sleep 50
sql connect
#todo wait for streamState checkpoint
return 1
#return 1
print =============== create database with retentions
sql create database d0 retentions -:7d,5m:21d,15m:365d;

18
tests/script/tsim/sync/vnodesnapshot-rsma-test.sim
View File

@ -114,7 +114,7 @@ endi
vg_ready:
print ====> create stable/child table
sql create table stb (ts timestamp, c1 int, c2 float, c3 double) tags (t1 int) rollup(sum) watermark 3s,3s max_delay 3s,3s
sql create table stb (ts timestamp, c1 float, c2 float, c3 double) tags (t1 int) rollup(sum) watermark 3s,3s max_delay 3s,3s
sql show stables
if $rows != 1 then
@ -167,9 +167,6 @@ system sh/exec.sh -n dnode4 -s start
sleep 3000
print =============== query data of level 1
sql connect
sql use db
@ -181,12 +178,21 @@ if $rows != 100 then
return -1
endi
print =============== sleep 5s to wait the result
sleep 5000
print =============== query data of level 2
sql select * from ct1 where ts > now - 10d
print rows of level 2: $rows
print $data00 $data01 $data02
print $data10 $data11 $data12
if $rows != 100 then
print rows of level 2: $rows
endi
print =============== query data of level 3
sql select * from ct1
print rows of level 3: $rows
print $data00 $data01 $data02
print $data10 $data11 $data12
if $rows != 100 then
print rows of level 3: $rows
endi

1
tests/script/win-test-file
View File

@ -320,6 +320,7 @@
./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
./test.sh -f tsim/sync/vnodesnapshot-rsma-test.sim
./test.sh -f tsim/valgrind/checkError1.sim
./test.sh -f tsim/valgrind/checkError2.sim
./test.sh -f tsim/valgrind/checkError3.sim

591
tests/system-test/0-others/view/non_marterial_view/test_view.py Normal file
View File

@ -0,0 +1,591 @@
import taos
import os
import sys
import time
from pathlib import Path
sys.path.append(os.path.dirname(Path(__file__).resolve().parent.parent.parent) + "/7-tmq")
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
from util.sqlset import *
from tmqCommon import *
class TDTestCase:
"""This test case is used to veirfy the tmq consume data from non marterial view
"""
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor())
self.setsql = TDSetSql()
# db info
self.dbname = "view_db"
self.stbname = 'stb'
self.ctbname_list = ["ct1", "ct2"]
self.stable_column_dict = {
'ts': 'timestamp',
'col1': 'float',
'col2': 'int',
}
self.tag_dict = {
'ctbname': 'binary(10)'
}
def prepare_data(self, conn=None):
"""Create the db and data for test
"""
tdLog.debug("Start to prepare the data")
if not conn:
conn = tdSql
# create datebase
conn.execute(f"create database {self.dbname}")
conn.execute(f"use {self.dbname}")
time.sleep(2)
# create stable
conn.execute(self.setsql.set_create_stable_sql(self.stbname, self.stable_column_dict, self.tag_dict))
tdLog.debug("Create stable {} successfully".format(self.stbname))
# create child tables
for ctname in self.ctbname_list:
conn.execute(f"create table {ctname} using {self.stbname} tags('{ctname}');")
tdLog.debug("Create child table {} successfully".format(ctname))
# insert data into child tables
conn.execute(f"insert into {ctname} values(now, 1.1, 1)(now+1s, 2.2, 2)(now+2s, 3.3, 3)(now+3s, 4.4, 4)(now+4s, 5.5, 5)(now+5s, 6.6, 6)(now+6s, 7.7, 7)(now+7s, 8.8, 8)(now+8s, 9.9, 9)(now+9s, 10.1, 10);)")
tdLog.debug(f"Insert into data to {ctname} successfully")
def prepare_tmq_data(self, para_dic):
tdLog.debug("Start to prepare the tmq data")
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, para_dic["dbName"], para_dic["dropFlag"], vgroups=para_dic["vgroups"], replica=1)
tdLog.info("create stb")
tdCom.create_stable(tdSql, dbname=para_dic["dbName"], stbname=para_dic["stbName"], column_elm_list=para_dic['colSchema'], tag_elm_list=para_dic['tagSchema'])
tdLog.info("create ctb")
tdCom.create_ctable(tdSql, dbname=para_dic["dbName"], stbname=para_dic["stbName"],tag_elm_list=para_dic['tagSchema'], count=para_dic["ctbNum"], default_ctbname_prefix=para_dic['ctbPrefix'])
tdLog.info("insert data")
tmqCom.insert_data(tdSql, para_dic["dbName"], para_dic["ctbPrefix"], para_dic["ctbNum"], para_dic["rowsPerTbl"], para_dic["batchNum"], para_dic["startTs"])
tdLog.debug("Finish to prepare the tmq data")
def check_view_num(self, num):
tdSql.query("show views;")
rows = tdSql.queryRows
assert(rows == num)
tdLog.debug(f"Verify the view number successfully")
def create_user(self, username, password):
tdSql.execute(f"create user {username} pass '{password}';")
tdLog.debug("Create user {} with password {} successfully".format(username, password))
def check_permissions(self, username, db_name, permission_dict, view_name=None):
"""
:param permission_dict: {'db': ["read", "write], 'view': ["read", "write", "alter"]}
"""
tdSql.query("select * from information_schema.ins_user_privileges;")
for item in permission_dict.keys():
if item == "db":
for permission in permission_dict[item]:
assert((username, permission, db_name, "", "", "") in tdSql.queryResult)
tdLog.debug(f"Verify the {item} {db_name} {permission} permission successfully")
elif item == "view":
for permission in permission_dict[item]:
assert((username, permission, db_name, view_name, "", "view") in tdSql.queryResult)
tdLog.debug(f"Verify the {item} {db_name} {view_name} {permission} permission successfully")
else:
raise Exception(f"Invalid permission type: {item}")
def test_create_view_from_one_database(self):
"""This test case is used to verify the create view from one database
"""
self.prepare_data()
tdSql.execute(f"create view v1 as select * from {self.stbname};")
self.check_view_num(1)
tdSql.error(f'create view v1 as select * from {self.stbname};', expectErrInfo='view already exists in db')
tdSql.error(f'create view db2.v2 as select * from {self.stbname};', expectErrInfo='Fail to get table info, error: Database not exist')
tdSql.error(f'create view v2 as select c2 from {self.stbname};', expectErrInfo='Invalid column name: c2')
tdSql.error(f'create view v2 as select ts, col1 from tt1;', expectErrInfo='Fail to get table info, error: Table does not exist')
tdSql.execute(f"drop database {self.dbname}")
tdLog.debug("Finish test case 'test_create_view_from_one_database'")
def test_create_view_from_multi_database(self):
"""This test case is used to verify the create view from multi database
"""
self.prepare_data()
tdSql.execute(f"create view v1 as select * from view_db.{self.stbname};")
self.check_view_num(1)
self.dbname = "view_db2"
self.prepare_data()
tdSql.execute(f"create view v1 as select * from view_db2.{self.stbname};")
tdSql.execute(f"create view v2 as select * from view_db.v1;")
self.check_view_num(2)
self.dbname = "view_db"
tdSql.execute(f"drop database view_db;")
tdSql.execute(f"drop database view_db2;")
tdLog.debug("Finish test case 'test_create_view_from_multi_database'")
def test_create_view_name_params(self):
"""This test case is used to verify the create view with different view name params
"""
self.prepare_data()
tdSql.execute(f"create view v1 as select * from {self.stbname};")
self.check_view_num(1)
tdSql.error(f"create view v/2 as select * from {self.stbname};", expectErrInfo='syntax error near "/2 as select * from stb;"')
tdSql.execute(f"create view v2 as select ts, col1 from {self.stbname};")
self.check_view_num(2)
view_name_192_characters = "rzuoxoIXilAGgzNjYActiQwgzZK7PZYpDuaOe1lSJMFMVYXaexh1OfMmk3LvJcQbTeXXW7uGJY8IHuweHF73VHgoZgf0waO33YpZiTKfDQbdWtN4YmR2eWjL84ZtkfjM4huCP6lCysbDMj8YNwWksTdUq70LIyNhHp2V8HhhxyYSkREYFLJ1kOE78v61MQT6"
tdSql.execute(f"create view {view_name_192_characters} as select * from {self.stbname};")
self.check_view_num(3)
tdSql.error(f"create view {view_name_192_characters}1 as select * from {self.stbname};", expectErrInfo='Invalid identifier name: rzuoxoixilaggznjyactiqwgzzk7pzypduaoe1lsjmfmvyxaexh1ofmmk3lvjcqbtexxw7ugjy8ihuwehf73vhgozgf0wao33ypzitkfdqbdwtn4ymr2ewjl84ztkfjm4hucp6lcysbdmj8ynwwkstduq70liynhhp2v8hhhxyyskreyflj1koe78v61mqt61 as select * from stb;')
tdSql.execute(f"drop database {self.dbname}")
tdLog.debug("Finish test case 'test_create_view_name_params'")
def test_create_view_query(self):
"""This test case is used to verify the create view with different data type in query
"""
self.prepare_data()
# add different data type table
tdSql.execute(f"create table tb (ts timestamp, c1 int, c2 int unsigned, c3 bigint, c4 bigint unsigned, c5 float, c6 double, c7 binary(16), c8 smallint, c9 smallint unsigned, c10 tinyint, c11 tinyint unsigned, c12 bool, c13 varchar(16), c14 nchar(8), c15 geometry(21), c16 varbinary(16));")
tdSql.execute(f"create view v1 as select ts, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14, c15, c16 from tb;")
# check data type in create view sql
tdSql.query("desc v1;")
res = tdSql.queryResult
data_type_list = [res[index][1] for index in range(len(res))]
tdLog.debug(data_type_list)
assert('TIMESTAMP' in data_type_list and 'INT' in data_type_list and 'INT UNSIGNED' in data_type_list and 'BIGINT' in data_type_list and 'BIGINT UNSIGNED' in data_type_list and 'FLOAT' in data_type_list and 'DOUBLE' in data_type_list and 'VARCHAR' in data_type_list and 'SMALLINT' in data_type_list and 'SMALLINT UNSIGNED' in data_type_list and 'TINYINT' in data_type_list and 'TINYINT UNSIGNED' in data_type_list and 'BOOL' in data_type_list and 'VARCHAR' in data_type_list and 'NCHAR' in data_type_list and 'GEOMETRY' in data_type_list and 'VARBINARY' in data_type_list)
tdSql.execute("create view v2 as select * from tb where c1 >5 and c7 like '%ab%';")
self.check_view_num(2)
tdSql.error("create view v3 as select * from tb where c1 like '%ab%';", expectErrInfo='Invalid value type')
tdSql.execute("create view v3 as select first(ts), sum(c1) from tb group by c2 having avg(c4) > 0;")
tdSql.execute("create view v4 as select _wstart,sum(c6) from tb interval(10s);")
tdSql.execute("create view v5 as select * from tb join v2 on tb.ts = v2.ts;")
tdSql.execute("create view v6 as select * from (select ts, c1, c2 from (select * from v2));")
self.check_view_num(6)
for v in ['v1', 'v2', 'v3', 'v4', 'v5', 'v6']:
tdSql.execute(f"drop view {v};")
tdSql.execute(f"drop database {self.dbname}")
tdLog.debug("Finish test case 'test_create_view_query'")
def test_show_view(self):
"""This test case is used to verify the show view
"""
self.prepare_data()
tdSql.execute(f"create view v1 as select * from {self.ctbname_list[0]};")
# query from show sql
tdSql.query("show views;")
res = tdSql.queryResult
assert(res[0][0] == 'v1' and res[0][1] == 'view_db' and res[0][2] == 'root' and res[0][4] == 'NORMAL' and res[0][5] == 'select * from ct1;')
# show create sql
tdSql.query("show create view v1;")
res = tdSql.queryResult
assert(res[0][1] == 'CREATE VIEW `view_db`.`v1` AS select * from ct1;')
# query from desc results
tdSql.query("desc view_db.v1;")
res = tdSql.queryResult
assert(res[0][1] == 'TIMESTAMP' and res[1][1] == 'FLOAT' and res[2][1] == 'INT')
# query from system table
tdSql.query("select * from information_schema.ins_views;")
res = tdSql.queryResult
assert(res[0][0] == 'v1' and res[0][1] == 'view_db' and res[0][2] == 'root' and res[0][4] == 'NORMAL' and res[0][5] == 'select * from ct1;')
tdSql.error("show db3.views;", expectErrInfo='Database not exist')
tdSql.error("desc viewx;", expectErrInfo='Table does not exist')
tdSql.error(f"show create view {self.dbname}.viewx;", expectErrInfo='view not exists in db')
tdSql.execute(f"drop database {self.dbname}")
tdSql.error("show views;", expectErrInfo='Database not exist')
tdLog.debug("Finish test case 'test_show_view'")
def test_drop_view(self):
"""This test case is used to verify the drop view
"""
self.prepare_data()
self.dbname = "view_db2"
self.prepare_data()
tdSql.execute("create view view_db.v1 as select * from view_db.stb;")
tdSql.execute("create view view_db2.v1 as select * from view_db2.stb;")
# delete view without database name
tdSql.execute("drop view v1;")
# delete view with database name
tdSql.execute("drop view view_db.v1;")
# delete non exist view
tdSql.error("drop view view_db.v11;", expectErrInfo='view not exists in db')
tdSql.execute("drop database view_db")
tdSql.execute("drop database view_db2;")
self.dbname = "view_db"
tdLog.debug("Finish test case 'test_drop_view'")
def test_view_permission_db_all_view_all(self):
"""This test case is used to verify the view permission with db all and view all,
the time sleep to wait the permission take effect
"""
self.prepare_data()
username = "view_test"
password = "test"
self.create_user(username, password)
# grant all db permission to user
tdSql.execute("grant all on view_db.* to view_test;")
conn = taos.connect(user=username, password=password)
conn.execute(f"use {self.dbname};")
conn.execute("create view v1 as select * from stb;")
res = conn.query("show views;")
assert(len(res.fetch_all()) == 1)
tdLog.debug(f"Verify the show view permission of user '{username}' with db all and view all successfully")
self.check_permissions("view_test", "view_db", {"db": ["read", "write"], "view": ["read", "write", "alter"]}, "v1")
tdLog.debug(f"Verify the view permission from system table successfully")
time.sleep(2)
conn.execute("drop view v1;")
tdSql.execute("revoke all on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname};")
time.sleep(1)
# prepare data by user 'view_test'
self.prepare_data(conn)
conn.execute("create view v1 as select * from stb;")
res = conn.query("show views;")
assert(len(res.fetch_all()) == 1)
tdLog.debug(f"Verify the view permission of user '{username}' with db all and view all successfully")
self.check_permissions("view_test", "view_db", {"db": ["read", "write"], "view": ["read", "write", "alter"]}, "v1")
tdLog.debug(f"Verify the view permission from system table successfully")
time.sleep(2)
conn.execute("drop view v1;")
tdSql.execute("revoke all on view_db.* from view_test;")
tdSql.execute("revoke all on view_db.v1 from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_all_view_all'")
def test_view_permission_db_write_view_all(self):
"""This test case is used to verify the view permission with db write and view all
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data(conn)
conn.execute("create view v1 as select * from stb;")
tdSql.execute("revoke read on view_db.* from view_test;")
self.check_permissions("view_test", "view_db", {"db": ["write"], "view": ["read", "write", "alter"]}, "v1")
# create view permission error
try:
conn.execute("create view v2 as select * from v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
# query from view permission error
try:
conn.query("select * from v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
# view query permission
res = conn.query("show views;")
assert(len(res.fetch_all()) == 1)
time.sleep(2)
conn.execute("drop view v1;")
tdSql.execute("revoke write on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_write_view_all'")
def test_view_permission_db_write_view_read(self):
"""This test case is used to verify the view permission with db write and view read
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data()
tdSql.execute("create view v1 as select * from stb;")
tdSql.execute("grant write on view_db.* to view_test;")
tdSql.execute("grant read on view_db.v1 to view_test;")
conn.execute(f"use {self.dbname};")
time.sleep(2)
res = conn.query("select * from v1;")
assert(len(res.fetch_all()) == 20)
conn.execute("create view v2 as select * from v1;")
# create view from super table of database
try:
conn.execute("create view v3 as select * from stb;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
time.sleep(2)
conn.execute("drop view v2;")
try:
conn.execute("drop view v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
tdSql.execute("revoke read on view_db.v1 from view_test;")
tdSql.execute("revoke write on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_write_view_read'")
def test_view_permission_db_write_view_alter(self):
"""This test case is used to verify the view permission with db write and view alter
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data()
tdSql.execute("create view v1 as select * from stb;")
tdSql.execute("grant write on view_db.* to view_test;")
tdSql.execute("grant alter on view_db.v1 to view_test;")
try:
conn.execute(f"use {self.dbname};")
conn.execute("select * from v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
time.sleep(2)
conn.execute("drop view v1;")
tdSql.execute("revoke write on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_write_view_alter'")
def test_view_permission_db_read_view_all(self):
"""This test case is used to verify the view permission with db read and view all
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data()
tdSql.execute("create view v1 as select * from stb;")
tdSql.execute("grant read on view_db.* to view_test;")
tdSql.execute("grant all on view_db.v1 to view_test;")
try:
conn.execute(f"use {self.dbname};")
conn.execute("create view v2 as select * from v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
time.sleep(2)
res = conn.query("select * from v1;")
assert(len(res.fetch_all()) == 20)
conn.execute("drop view v1;")
tdSql.execute("revoke read on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_read_view_all'")
def test_view_permission_db_read_view_alter(self):
"""This test case is used to verify the view permission with db read and view alter
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data()
tdSql.execute("create view v1 as select * from stb;")
tdSql.execute("grant read on view_db.* to view_test;")
tdSql.execute("grant alter on view_db.v1 to view_test;")
try:
conn.execute(f"use {self.dbname};")
conn.execute("select * from v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
time.sleep(2)
conn.execute("drop view v1;")
tdSql.execute("revoke read on view_db.* from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_read_view_alter'")
def test_view_permission_db_read_view_read(self):
"""This test case is used to verify the view permission with db read and view read
"""
username = "view_test"
password = "test"
self.create_user(username, password)
conn = taos.connect(user=username, password=password)
self.prepare_data()
tdSql.execute("create view v1 as select * from stb;")
tdSql.execute("grant read on view_db.* to view_test;")
tdSql.execute("grant read on view_db.v1 to view_test;")
conn.execute(f"use {self.dbname};")
time.sleep(2)
res = conn.query("select * from v1;")
assert(len(res.fetch_all()) == 20)
try:
conn.execute("drop view v1;")
except Exception as ex:
assert("[0x2644]: Permission denied or target object not exist" in str(ex))
tdSql.execute("revoke read on view_db.* from view_test;")
tdSql.execute("revoke read on view_db.v1 from view_test;")
tdSql.execute(f"drop database {self.dbname}")
tdSql.execute("drop user view_test;")
tdLog.debug("Finish test case 'test_view_permission_db_read_view_read'")
def test_query_from_view(self):
"""This test case is used to verify the query from view
"""
self.prepare_data()
view_name_list = []
# common query from super table
tdSql.execute(f"create view v1 as select * from {self.stbname};")
tdSql.query(f"select * from v1;")
rows = tdSql.queryRows
assert(rows == 20)
view_name_list.append("v1")
tdLog.debug("Verify the query from super table successfully")
# common query from child table
tdSql.execute(f"create view v2 as select * from {self.ctbname_list[0]};")
tdSql.query(f"select * from v2;")
rows = tdSql.queryRows
assert(rows == 10)
view_name_list.append("v2")
tdLog.debug("Verify the query from child table successfully")
# join query
tdSql.execute(f"create view v3 as select * from {self.stbname} join {self.ctbname_list[1]} on {self.ctbname_list[1]}.ts = {self.stbname}.ts;")
tdSql.query(f"select * from v3;")
rows = tdSql.queryRows
assert(rows == 10)
view_name_list.append("v3")
tdLog.debug("Verify the join query successfully")
# group by query
tdSql.execute(f"create view v4 as select count(*) from {self.stbname} group by tbname;")
tdSql.query(f"select * from v4;")
rows = tdSql.queryRows
assert(rows == 2)
res = tdSql.queryResult
assert(res[0][0] == 10)
view_name_list.append("v4")
tdLog.debug("Verify the group by query successfully")
# partition by query
tdSql.execute(f"create view v5 as select sum(col1) from {self.stbname} where col2 > 4 partition by tbname interval(3s);")
tdSql.query(f"select * from v5;")
rows = tdSql.queryRows
assert(rows >= 4)
view_name_list.append("v5")
tdLog.debug("Verify the partition by query successfully")
# query from nested view
tdSql.execute(f"create view v6 as select * from v5;")
tdSql.query(f"select * from v6;")
rows = tdSql.queryRows
assert(rows >= 4)
view_name_list.append("v6")
tdLog.debug("Verify the query from nested view successfully")
# delete view
for view in view_name_list:
tdSql.execute(f"drop view {view};")
tdLog.debug(f"Drop view {view} successfully")
tdSql.execute(f"drop database {self.dbname}")
tdLog.debug("Finish test case 'test_query_from_view'")
def test_tmq_from_view(self):
"""This test case is used to verify the tmq consume data from view
"""
# params for db
paraDict = {'dbName': 'view_db',
'dropFlag': 1,
'event': '',
'vgroups': 4,
'stbName': 'stb',
'colPrefix': 'c',
'tagPrefix': 't',
'colSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
'tagSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
'ctbPrefix': 'ctb',
'ctbNum': 1,
'rowsPerTbl': 10000,
'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 10,
'showMsg': 1,
'showRow': 1}
# topic info
topic_name_list = ['topic1']
view_name_list = ['view1']
expectRowsList = []
self.prepare_tmq_data(paraDict)
# init consume info, and start tmq_sim, then check consume result
tmqCom.initConsumerTable()
queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
tdSql.execute(f"create view {view_name_list[0]} as {queryString}")
sqlString = "create topic %s as %s" %(topic_name_list[0], "select * from %s"%view_name_list[0])
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
tdSql.query(queryString)
expectRowsList.append(tdSql.getRows())
consumerId = 1
topicList = topic_name_list[0]
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
keyList = 'group.id:cgrp1, enable.auto.commit:false, auto.commit.interval.ms:6000, auto.offset.reset:earliest'
ifcheckdata = 1
ifManualCommit = 1
tmqCom.insertConsumerInfo(consumerId, expectrowcnt, topicList, keyList, ifcheckdata, ifManualCommit)
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(paraDict['pollDelay'], paraDict["dbName"], paraDict['showMsg'], paraDict['showRow'])
tdLog.info("wait the consume result")
expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows)
if expectRowsList[0] != resultList[0]:
tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0]))
tdLog.exit("1 tmq consume rows error!")
tmqCom.checkFileContent(consumerId, queryString)
time.sleep(10)
for i in range(len(topic_name_list)):
tdSql.query("drop topic %s"%topic_name_list[i])
for i in range(len(view_name_list)):
tdSql.query("drop view %s"%view_name_list[i])
# drop database
tdSql.execute(f"drop database {paraDict['dbName']}")
tdSql.execute("drop database cdb;")
tdLog.debug("Finish test case 'test_tmq_from_view'")
def run(self):
self.test_create_view_from_one_database()
self.test_create_view_from_multi_database()
self.test_create_view_name_params()
self.test_create_view_query()
self.test_show_view()
self.test_drop_view()
self.test_view_permission_db_all_view_all()
self.test_view_permission_db_write_view_all()
self.test_view_permission_db_write_view_read()
self.test_view_permission_db_write_view_alter()
self.test_view_permission_db_read_view_all()
self.test_view_permission_db_read_view_alter()
self.test_view_permission_db_read_view_read()
self.test_query_from_view()
self.test_tmq_from_view()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())