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

Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/syntax

This commit is contained in:
54liuyao 2024-10-12 15:02:55 +08:00
parent 50ceb19cbf c44c8ec106
commit 10810d48de
46 changed files with 624 additions and 249 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored
View File

@ -138,3 +138,24 @@ tags
*CMakeCache*
*CMakeFiles*
.history/
*.txt
*.tcl
*.pc
contrib/geos
contrib/libuv
contrib/pcre2
contrib/zlib
deps_tmp_CMakeLists.txt.in
*.a
*.ctest
pcre2-config
pcre2_test.sh
pcre2_grep_test.sh
pcre2_chartables.c
geos-config
config.h
pcre2.h
zconf.h
version.h
geos_c.h

2
cmake/stub_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# stub
ExternalProject_Add(stub
GIT_REPOSITORY https://github.com/coolxv/cpp-stub.git
GIT_TAG 5e903b8e
GIT_TAG 3137465194014d66a8402941e80d2bccc6346f51
GIT_SUBMODULES "src"
SOURCE_DIR "${TD_CONTRIB_DIR}/cpp-stub"
BINARY_DIR "${TD_CONTRIB_DIR}/cpp-stub/src"

11
docs/zh/14-reference/03-taos-sql/24-show.md
View File

@ -30,6 +30,16 @@ SHOW CLUSTER ALIVE;
查询当前集群的状态是否可用，返回值： 0：不可用 1：完全可用 2：部分可用（集群中部分节点下线，但其它节点仍可以正常使用）
## SHOW CLUSTER MACHINES
```sql
SHOW CLUSTER MACHINES; // 从 TDengine 3.2.3.0 版本开始支持
```
显示集群的机器码等信息。
注：企业版独有
## SHOW CONNECTIONS
```sql
@ -99,6 +109,7 @@ SHOW FUNCTIONS;
```sql
SHOW LICENCES;
SHOW GRANTS;
SHOW GRANTS FULL; // 从 TDengine 3.2.3.0 版本开始支持
```
显示企业版许可授权的信息。

89
docs/zh/28-releases/03-notes/3.3.3.0.md Normal file
View File

@ -0,0 +1,89 @@
---
title: 3.3.3.0 版本说明
sidebar_label: 3.3.3.0
description: 3.3.3.0 版本说明
---
### 新特性/优化
1. 增加函数支持：pi、truncate/tunc、exp、ln、mod、rand、sign、degress、radians、char、ascii、position、trim、replace、repeat、substring、substr、substring_index、week、weekday、weekofyear、dayofweek、stddev_pop、var_pop
2. 多级存储支持微软对象存储 Azure Blob [企业版]
3. 支持 MongoDB 数据源 [企业版]
4. TDengine支持macOS企业版客户端 [企业版]
5. taosX日志默认不写入syslog [企业版]
6. 服务端记录所有慢查询信息到log库
7. show cluster machines 查询结果中添加服务端版本号
8. 删除保留关键字LEVEL/ENCODE/COMPRESS, 可以作为列名/表名/数据库名等使用
9. 禁止动态修改临时目录
10. round 函数：支持四舍五入的精度
11. timediff 函数：返回时间戳 expr1 - expr2 的结果，支持负数结果，并近似到时间单位 time_unit 指定的精度
12. max/min 函数：支持字符串作为输入参数，当输入参数为字符串类型时，返回最大字符串值
13. 提升了 taosX 数据同步的性能 [企业版]
14. 统一慢查询和普通日志的保存策略
15. 优化订阅功能中offset文件过大且加载过慢的问题
16. 升级 JDBC driver 至 3.2.5
17. Kafka connector 提交改为同步方式
18. 新增参数：表查询不存在时是否报异常
19. 扩大 tsma 的 interval 的最大窗口到 1 年
20. 支持从 CSV 文件批量建表
### 修复问题
1. 服务器重启后 mqtt 写入任务未自动重新启动 [企业版]
2. 修复windows上agent和PI connector连接错误导致任务终止的问题 [企业版]
3. 重启taosadapter后,mqtt 任务异常，无法同步数据 [企业版]
4. Explorer 上 mqtt 写入任务状态不正确 [企业版]
5. 重启 taosx 后，运行中的任务没有被重新调度 [企业版]
6. 对于 taosx 的 MongoDB DataIn 任务，优化了日志的输出信息 [企业版]
7. 修复在 local.toml 文件为空时备份失败的问题 [企业版]
8. taosx replica 任务在网络断开恢复后数据同步异常的问题 [企业版]
9. 修复3.1版本同步到3.3版本meta数据不同步问题 [企业版]
10. taosadapter 异常重启后, mqtt任务无法自动恢复的问题[企业版]
11. Kafka DataIn 任务状态正常，但任务停止消费数据 [企业版]
12. 修复 last + interval 查询导致 crash 的问题
13. 在数据写入时热更stt_trigger导致taosd崩溃
14. 修改root密码后taoskeeper反复重启
15. 云服务中服务重启后流计算无法继续
16. tsdb加载缓存导致taosd死锁
17. 查询时fill值指定为0报错
18. last_row 查询结果与预期不符
19. 用户指定 information_schema 库登录时导致 taosd 崩溃
20. group by 语句支持位置参数和别名语法
21. 元数据克隆函数对象错误导致 crash
22. 缓存更新时数据填充的游标错误导致的 taosd 异常退出
23. STDDEV 函数计算结果随机不正确的问题
24. 多级存储以及加密场景下无法添加离线节点
25. taos CLI 无法输入大于20字节长度的密码
26. 拼接 sql 写入报错: int data overflow
27. 大量查询并发场景下元数据的一致性
28. 尝试解决手动点击停止按钮，任务无法停止的问题
29. 解决了列或者标签中包含逗号或引号时导出 CSV 文件错误的问题
30. 多线程并发的对同一张子表进行 describe 导致 crash
31. 数据浏览器的查询结果未按照配置的时区展示的问题
32. 创建 websocket 连接时 taosadapter 内存泄漏的问题
33. 社区版无法在 redhat 操作系统正常启动
34. 无效断言在两个 stt 分别位于数据块时间轴分布空洞场景引发的 crash
35. S3 长时间拉取数据会失败的问题
36. 使用结果集函数判断时间线的有序性
37. 解决超级表投影查询慢的问题
38. 修复: 多节点环境，事务返回码与事务状态不一致导致事务堆积
39. 修复：在stt_trigger = 1时，delete数据之后重新出现的错误
40. 节点恢复阶段taosd内存占用过高导致OOM
41. limit过小时的判断错误
42. AVEVA historian 数据源连通性及获取示例数据功能
43. taosd 服务无法正常停止的问题
44. last 函数查询全空列导致 crash
45. 查询 cid 不在当前版本的 schema 时触发 assert 导致 coredump
46. 写入消息体过大时 OOM 的问题，增加 syncLogBufferMemoryAllowed 参数控制
47. timezone 设置为 UTC 且数据包含“1970”时,查询报错
48. 取消mnode节点的消息同步的等待超时
49. 集群节点异常恢复后，集群授权状态失效
50. 查询空指针问题
51. 在restore vnode时，将vgroup暂时设置成单副本
52. 可以使用SQL保留字作为列名创建超级表，也可以使用SQL保留字作为标签名
53. 高并发查询期间内存持续增长的问题
54. 流计算在checkpoint处理过程中，由于断言失败导致的crash
55. 修复了 3.3.2.8.0820 版本中可以查询到 information_schema 中不存在的表perf_queries问题
56. 节点恢复阶段taosd内存占用过高导致OOM
57. 修复在schema变更情况下，通过taosx导入数据时，taosd出现crash的问题
58. 可能的内存泄漏
59. 客户端内存泄漏
60. 开源用户修改stt_trigger值升级后无法修改其他数据库选项
61. NOT IN (NULL) 查询结果不正确
62. taos shell和taosBenchmark不能成功连接云服务实例

2
docs/zh/28-releases/03-notes/index.md
View File

@ -3,5 +3,5 @@ title: 版本说明
sidebar_label: 版本说明
description: 各版本版本说明
---
[3.3.3.0](./3.3.3.0)
[3.3.2.0](./3.3.2.0)

1
include/client/taos.h
View File

@ -251,6 +251,7 @@ DLL_EXPORT int64_t taos_affected_rows64(TAOS_RES *res);
DLL_EXPORT TAOS_FIELD *taos_fetch_fields(TAOS_RES *res);
DLL_EXPORT int taos_select_db(TAOS *taos, const char *db);
DLL_EXPORT int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields);
DLL_EXPORT int taos_print_row_with_size(char *str, uint32_t size, TAOS_ROW row, TAOS_FIELD *fields, int num_fields);
DLL_EXPORT void taos_stop_query(TAOS_RES *res);
DLL_EXPORT bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col);
DLL_EXPORT int taos_is_null_by_column(TAOS_RES *res, int columnIndex, bool result[], int *rows);

2
include/common/tdatablock.h
View File

@ -277,7 +277,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pData
bool alreadyAddGroupId(char* ctbName, int64_t groupId);
bool isAutoTableName(char* ctbName);
void buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId);
int32_t buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId, size_t cap);
int32_t buildCtbNameByGroupId(const char* stbName, uint64_t groupId, char** pName);
int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);

2
include/libs/nodes/cmdnodes.h
View File

@ -320,7 +320,7 @@ typedef struct SAlterDnodeStmt {
typedef struct {
ENodeType type;
char url[TSDB_ANAL_ANODE_URL_LEN];
char url[TSDB_ANAL_ANODE_URL_LEN + 3];
} SCreateAnodeStmt;
typedef struct {

7
include/util/taoserror.h
View File

@ -482,9 +482,10 @@ int32_t taosGetErrSize();
#define TSDB_CODE_MND_ANODE_TOO_LONG_URL TAOS_DEF_ERROR_CODE(0, 0x0432)
#define TSDB_CODE_MND_ANODE_INVALID_PROTOCOL TAOS_DEF_ERROR_CODE(0, 0x0433)
#define TSDB_CODE_MND_ANODE_INVALID_VERSION TAOS_DEF_ERROR_CODE(0, 0x0434)
#define TSDB_CODE_MND_ANODE_TOO_MANY_ALGO TAOS_DEF_ERROR_CODE(0, 0x0435)
#define TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME TAOS_DEF_ERROR_CODE(0, 0x0436)
#define TSDB_CODE_MND_ANODE_TOO_MANY_ALGO_TYPE TAOS_DEF_ERROR_CODE(0, 0x0437)
#define TSDB_CODE_MND_ANODE_INVALID_ALGO_TYPE TAOS_DEF_ERROR_CODE(0, 0x0435)
#define TSDB_CODE_MND_ANODE_TOO_MANY_ALGO TAOS_DEF_ERROR_CODE(0, 0x0436)
#define TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME TAOS_DEF_ERROR_CODE(0, 0x0437)
#define TSDB_CODE_MND_ANODE_TOO_MANY_ALGO_TYPE TAOS_DEF_ERROR_CODE(0, 0x0438)
// analysis
#define TSDB_CODE_ANAL_URL_RSP_IS_NULL TAOS_DEF_ERROR_CODE(0, 0x0440)

1
include/util/tjson.h
View File

@ -71,6 +71,7 @@ int32_t tjsonGetObjectValueString(const SJson* pJson, char** pStringValue);
void tjsonGetObjectValueBigInt(const SJson* pJson, int64_t* pVal);
void tjsonGetObjectValueDouble(const SJson* pJson, double* pVal);
int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal);
int32_t tjsonGetStringValue2(const SJson* pJson, const char* pName, char* pVal, int32_t maxLen);
int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal);
int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal);
int32_t tjsonGetIntValue(const SJson* pJson, const char* pName, int32_t* pVal);

12
source/client/src/clientEnv.c
View File

@ -114,7 +114,7 @@ static void concatStrings(SArray *list, char *buf, int size) {
db = dot + 1;
}
if (i != 0) {
(void)strcat(buf, ",");
(void)strncat(buf, ",", size - 1 - len);
len += 1;
}
int ret = snprintf(buf + len, size - len, "%s", db);
@ -1132,27 +1132,27 @@ static setConfRet taos_set_config_imp(const char *config){
static bool setConfFlag = false;
if (setConfFlag) {
ret.retCode = SET_CONF_RET_ERR_ONLY_ONCE;
strcpy(ret.retMsg, "configuration can only set once");
tstrncpy(ret.retMsg, "configuration can only set once", RET_MSG_LENGTH);
return ret;
}
taosInitGlobalCfg();
cJSON *root = cJSON_Parse(config);
if (root == NULL){
ret.retCode = SET_CONF_RET_ERR_JSON_PARSE;
strcpy(ret.retMsg, "parse json error");
tstrncpy(ret.retMsg, "parse json error", RET_MSG_LENGTH);
return ret;
}
int size = cJSON_GetArraySize(root);
if(!cJSON_IsObject(root) || size == 0) {
ret.retCode = SET_CONF_RET_ERR_JSON_INVALID;
strcpy(ret.retMsg, "json content is invalid, must be not empty object");
tstrncpy(ret.retMsg, "json content is invalid, must be not empty object", RET_MSG_LENGTH);
return ret;
}
if(size >= 1000) {
ret.retCode = SET_CONF_RET_ERR_TOO_LONG;
strcpy(ret.retMsg, "json object size is too long");
tstrncpy(ret.retMsg, "json object size is too long", RET_MSG_LENGTH);
return ret;
}
@ -1160,7 +1160,7 @@ static setConfRet taos_set_config_imp(const char *config){
cJSON *item = cJSON_GetArrayItem(root, i);
if(!item) {
ret.retCode = SET_CONF_RET_ERR_INNER;
strcpy(ret.retMsg, "inner error");
tstrncpy(ret.retMsg, "inner error", RET_MSG_LENGTH);
return ret;
}
if(!taosReadConfigOption(item->string, item->valuestring, NULL, NULL, TAOS_CFG_CSTATUS_OPTION, TSDB_CFG_CTYPE_B_CLIENT)){

2
source/client/src/clientHb.c
View File

@ -772,7 +772,7 @@ static int32_t hbGetUserAuthInfo(SClientHbKey *connKey, SHbParam *param, SClient
SUserAuthVersion *qUserAuth =
(SUserAuthVersion *)taosMemoryRealloc(pKv->value, (userNum + 1) * sizeof(SUserAuthVersion));
if (qUserAuth) {
(void)strncpy((qUserAuth + userNum)->user, pTscObj->user, TSDB_USER_LEN);
tstrncpy((qUserAuth + userNum)->user, pTscObj->user, TSDB_USER_LEN);
(qUserAuth + userNum)->version = htonl(-1); // force get userAuthInfo
pKv->value = qUserAuth;
pKv->valueLen += sizeof(SUserAuthVersion);

38
source/client/src/clientImpl.c
View File

@ -1921,19 +1921,19 @@ TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, cons
return NULL;
}
TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, const char* pass, int passLen,
const char* db, int dbLen, uint16_t port) {
char ipStr[TSDB_EP_LEN] = {0};
char dbStr[TSDB_DB_NAME_LEN] = {0};
char userStr[TSDB_USER_LEN] = {0};
char passStr[TSDB_PASSWORD_LEN] = {0};
(void)strncpy(ipStr, ip, TMIN(TSDB_EP_LEN - 1, ipLen));
(void)strncpy(userStr, user, TMIN(TSDB_USER_LEN - 1, userLen));
(void)strncpy(passStr, pass, TMIN(TSDB_PASSWORD_LEN - 1, passLen));
(void)strncpy(dbStr, db, TMIN(TSDB_DB_NAME_LEN - 1, dbLen));
return taos_connect(ipStr, userStr, passStr, dbStr, port);
}
//TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, const char* pass, int passLen,
// const char* db, int dbLen, uint16_t port) {
// char ipStr[TSDB_EP_LEN] = {0};
// char dbStr[TSDB_DB_NAME_LEN] = {0};
// char userStr[TSDB_USER_LEN] = {0};
// char passStr[TSDB_PASSWORD_LEN] = {0};
//
// tstrncpy(ipStr, ip, TMIN(TSDB_EP_LEN - 1, ipLen));
// tstrncpy(userStr, user, TMIN(TSDB_USER_LEN - 1, userLen));
// tstrncpy(passStr, pass, TMIN(TSDB_PASSWORD_LEN - 1, passLen));
// tstrncpy(dbStr, db, TMIN(TSDB_DB_NAME_LEN - 1, dbLen));
// return taos_connect(ipStr, userStr, passStr, dbStr, port);
//}
void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
for (int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
@ -2275,7 +2275,7 @@ static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int
char* jsonInnerData = data + CHAR_BYTES;
char dst[TSDB_MAX_JSON_TAG_LEN] = {0};
if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
(void)sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
(void)snprintf(varDataVal(dst), TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE, "%s", TSDB_DATA_NULL_STR_L);
varDataSetLen(dst, strlen(varDataVal(dst)));
} else if (tTagIsJson(data)) {
char* jsonString = NULL;
@ -2298,10 +2298,10 @@ static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int
*(char*)POINTER_SHIFT(varDataVal(dst), length + CHAR_BYTES) = '\"';
} else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
double jsonVd = *(double*)(jsonInnerData);
(void)sprintf(varDataVal(dst), "%.9lf", jsonVd);
(void)snprintf(varDataVal(dst), TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE, "%.9lf", jsonVd);
varDataSetLen(dst, strlen(varDataVal(dst)));
} else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) {
(void)sprintf(varDataVal(dst), "%s", (*((char*)jsonInnerData) == 1) ? "true" : "false");
(void)snprintf(varDataVal(dst), TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE, "%s", (*((char*)jsonInnerData) == 1) ? "true" : "false");
varDataSetLen(dst, strlen(varDataVal(dst)));
} else {
tscError("doConvertJson error: invalid type:%d", jsonInnerType);
@ -2658,8 +2658,8 @@ int32_t appendTbToReq(SHashObj* pHash, int32_t pos1, int32_t len1, int32_t pos2,
return -1;
}
char dbFName[TSDB_DB_FNAME_LEN];
(void)sprintf(dbFName, "%d.%.*s", acctId, dbLen, dbName);
char dbFName[TSDB_DB_FNAME_LEN] = {0};
(void)snprintf(dbFName, TSDB_DB_FNAME_LEN, "%d.%.*s", acctId, dbLen, dbName);
STablesReq* pDb = taosHashGet(pHash, dbFName, strlen(dbFName));
if (pDb) {
@ -2672,7 +2672,7 @@ int32_t appendTbToReq(SHashObj* pHash, int32_t pos1, int32_t len1, int32_t pos2,
if (NULL == db.pTables) {
return terrno;
}
(void)strcpy(db.dbFName, dbFName);
tstrncpy(db.dbFName, dbFName, TSDB_DB_FNAME_LEN);
if (NULL == taosArrayPush(db.pTables, &name)) {
return terrno;
}

4
source/client/src/clientJniConnector.c
View File

@ -1083,14 +1083,14 @@ JNIEXPORT jstring JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_stmtErrorMsgIm
TAOS *tscon = (TAOS *)con;
if (tscon == NULL) {
jniError("jobj:%p, connection already closed", jobj);
(void)sprintf(errMsg, "jobj:%p, connection already closed", jobj);
(void)snprintf(errMsg, sizeof(errMsg), "jobj:%p, connection already closed", jobj);
return (*env)->NewStringUTF(env, errMsg);
}
TAOS_STMT *pStmt = (TAOS_STMT *)stmt;
if (pStmt == NULL) {
jniError("jobj:%p, conn:%p, invalid stmt", jobj, tscon);
(void)sprintf(errMsg, "jobj:%p, conn:%p, invalid stmt", jobj, tscon);
(void)snprintf(errMsg, sizeof(errMsg), "jobj:%p, conn:%p, invalid stmt", jobj, tscon);
return (*env)->NewStringUTF(env, errMsg);
}

55
source/client/src/clientMain.c
View File

@ -482,71 +482,75 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
}
int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
return taos_print_row_with_size(str, INT32_MAX, row, fields, num_fields);
}
int taos_print_row_with_size(char *str, uint32_t size, TAOS_ROW row, TAOS_FIELD *fields, int num_fields){
int32_t len = 0;
for (int i = 0; i < num_fields; ++i) {
if (i > 0) {
if (i > 0 && len < size - 1) {
str[len++] = ' ';
}
if (row[i] == NULL) {
len += sprintf(str + len, "%s", TSDB_DATA_NULL_STR);
len += snprintf(str + len, size - len, "%s", TSDB_DATA_NULL_STR);
continue;
}
switch (fields[i].type) {
case TSDB_DATA_TYPE_TINYINT:
len += sprintf(str + len, "%d", *((int8_t *)row[i]));
len += snprintf(str + len, size - len, "%d", *((int8_t *)row[i]));
break;
case TSDB_DATA_TYPE_UTINYINT:
len += sprintf(str + len, "%u", *((uint8_t *)row[i]));
len += snprintf(str + len, size - len, "%u", *((uint8_t *)row[i]));
break;
case TSDB_DATA_TYPE_SMALLINT:
len += sprintf(str + len, "%d", *((int16_t *)row[i]));
len += snprintf(str + len, size - len, "%d", *((int16_t *)row[i]));
break;
case TSDB_DATA_TYPE_USMALLINT:
len += sprintf(str + len, "%u", *((uint16_t *)row[i]));
len += snprintf(str + len, size - len, "%u", *((uint16_t *)row[i]));
break;
case TSDB_DATA_TYPE_INT:
len += sprintf(str + len, "%d", *((int32_t *)row[i]));
len += snprintf(str + len, size - len, "%d", *((int32_t *)row[i]));
break;
case TSDB_DATA_TYPE_UINT:
len += sprintf(str + len, "%u", *((uint32_t *)row[i]));
len += snprintf(str + len, size - len, "%u", *((uint32_t *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
len += snprintf(str + len, size - len, "%" PRId64, *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_UBIGINT:
len += sprintf(str + len, "%" PRIu64, *((uint64_t *)row[i]));
len += snprintf(str + len, size - len, "%" PRIu64, *((uint64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT: {
float fv = 0;
fv = GET_FLOAT_VAL(row[i]);
len += sprintf(str + len, "%f", fv);
len += snprintf(str + len, size - len, "%f", fv);
} break;
case TSDB_DATA_TYPE_DOUBLE: {
double dv = 0;
dv = GET_DOUBLE_VAL(row[i]);
len += sprintf(str + len, "%lf", dv);
len += snprintf(str + len, size - len, "%lf", dv);
} break;
case TSDB_DATA_TYPE_VARBINARY: {
void *data = NULL;
uint32_t size = 0;
uint32_t tmp = 0;
int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
if (taosAscii2Hex(row[i], charLen, &data, &size) < 0) {
if (taosAscii2Hex(row[i], charLen, &data, &tmp) < 0) {
break;
}
(void)memcpy(str + len, data, size);
len += size;
uint32_t copyLen = TMIN(size - len - 1, tmp);
(void)memcpy(str + len, data, copyLen);
len += copyLen;
taosMemoryFree(data);
} break;
case TSDB_DATA_TYPE_BINARY:
@ -566,21 +570,28 @@ int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields)
}
}
(void)memcpy(str + len, row[i], charLen);
len += charLen;
uint32_t copyLen = TMIN(size - len - 1, charLen);
(void)memcpy(str + len, row[i], copyLen);
len += copyLen;
} break;
case TSDB_DATA_TYPE_TIMESTAMP:
len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
len += snprintf(str + len, size - len, "%" PRId64, *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_BOOL:
len += sprintf(str + len, "%d", *((int8_t *)row[i]));
len += snprintf(str + len, size - len, "%d", *((int8_t *)row[i]));
default:
break;
}
if (len >= size - 1) {
break;
}
}
if (len < size){
str[len] = 0;
}
str[len] = 0;
return len;
}
@ -945,7 +956,7 @@ int taos_get_current_db(TAOS *taos, char *database, int len, int *required) {
if (required) *required = strlen(pTscObj->db) + 1;
TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
} else {
(void)strcpy(database, pTscObj->db);
tstrncpy(database, pTscObj->db, len);
code = 0;
}
_return:

6
source/client/src/clientMonitor.c
View File

@ -183,7 +183,7 @@ FAILED:
static void generateClusterReport(taos_collector_registry_t* registry, void* pTransporter, SEpSet* epSet) {
char ts[50] = {0};
(void)sprintf(ts, "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI));
(void)snprintf(ts, sizeof(ts), "%" PRId64, taosGetTimestamp(TSDB_TIME_PRECISION_MILLI));
char* pCont = (char*)taos_collector_registry_bridge_new(registry, ts, "%" PRId64, NULL);
if (NULL == pCont) {
tscError("generateClusterReport failed, get null content.");
@ -401,7 +401,7 @@ static void monitorWriteSlowLog2File(MonitorSlowLogData* slowLogData, char* tmpP
return;
}
pClient->lastCheckTime = taosGetMonoTimestampMs();
(void)strcpy(pClient->path, path);
tstrncpy(pClient->path, path, PATH_MAX);
pClient->offset = 0;
pClient->pFile = pFile;
if (taosHashPut(monitorSlowLogHash, &slowLogData->clusterId, LONG_BYTES, &pClient, POINTER_BYTES) != 0) {
@ -458,7 +458,7 @@ static char* readFile(TdFilePtr pFile, int64_t* offset, int64_t size) {
return NULL;
}
char* buf = pCont;
(void)strcat(buf++, "[");
(void)strncat(buf++, "[", totalSize - 1);
int64_t readSize = taosReadFile(pFile, buf, totalSize - 4); // 4 reserved for []
if (readSize <= 0) {
if (readSize < 0) {

16
source/client/src/clientRawBlockWrite.c
View File

@ -926,7 +926,7 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) {
for (int32_t i = 0; i < req.schemaRow.nCols; i++) {
SSchema* pSchema = req.schemaRow.pSchema + i;
SFieldWithOptions field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};
(void)strcpy(field.name, pSchema->name);
tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);
if (createDefaultCompress) {
field.compress = createDefaultColCmprByType(pSchema->type);
@ -941,7 +941,7 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) {
for (int32_t i = 0; i < req.schemaTag.nCols; i++) {
SSchema* pSchema = req.schemaTag.pSchema + i;
SField field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};
(void)strcpy(field.name, pSchema->name);
tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);
RAW_NULL_CHECK(taosArrayPush(pReq.pTags, &field));
}
@ -1244,7 +1244,7 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
if (pTableBatch == NULL) {
SVgroupCreateTableBatch tBatch = {0};
tBatch.info = pInfo;
(void)strcpy(tBatch.dbName, pRequest->pDb);
tstrncpy(tBatch.dbName, pRequest->pDb, TSDB_DB_NAME_LEN);
tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));
RAW_NULL_CHECK(tBatch.req.pArray);
@ -1769,8 +1769,8 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
RAW_NULL_CHECK(tbName);
SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};
(void)strcpy(pName.dbname, pRequest->pDb);
(void)strcpy(pName.tname, tbName);
tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);
tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);
RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));
@ -1928,15 +1928,15 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
uDebug(LOG_ID_TAG " write raw metadata block tbname:%s", LOG_ID_VALUE, tbName);
SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};
(void)strcpy(pName.dbname, pRequest->pDb);
(void)strcpy(pName.tname, tbName);
tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);
tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);
// find schema data info
SVCreateTbReq* pCreateReqDst = (SVCreateTbReq*)taosHashGet(pCreateTbHash, tbName, strlen(tbName));
SVgroupInfo vg = {0};
RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vg));
if (pCreateReqDst) { // change stable name to get meta
(void)strcpy(pName.tname, pCreateReqDst->ctb.stbName);
tstrncpy(pName.tname, pCreateReqDst->ctb.stbName, TSDB_TABLE_NAME_LEN);
}
RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));

9
source/client/src/clientSml.c
View File

@ -137,7 +137,7 @@ void smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2
}
(void)memset(pBuf->buf, 0, pBuf->len);
if (msg1) {
(void)strncat(pBuf->buf, msg1, pBuf->len);
(void)strncat(pBuf->buf, msg1, pBuf->len - 1);
}
int32_t left = pBuf->len - strlen(pBuf->buf);
if (left > 2 && msg2) {
@ -515,9 +515,9 @@ static int32_t smlParseTableName(SArray *tags, char *childTableName, char *tbnam
if (tag == NULL) {
return TSDB_CODE_SML_INVALID_DATA;
}
(void)strncat(childTableName, tag->value, tag->length);
(void)strncat(childTableName, tag->value, TMIN(tag->length, TSDB_TABLE_NAME_LEN - 1 - strlen(childTableName)));
if (i != taosArrayGetSize(tags) - 1) {
(void)strcat(childTableName, tsSmlAutoChildTableNameDelimiter);
(void)strncat(childTableName, tsSmlAutoChildTableNameDelimiter, TSDB_TABLE_NAME_LEN - 1 - strlen(childTableName));
}
}
if (tsSmlDot2Underline) {
@ -538,8 +538,7 @@ static int32_t smlParseTableName(SArray *tags, char *childTableName, char *tbnam
// handle child table name
if (childTableNameLen == tag->keyLen && strncmp(tag->key, tbnameKey, tag->keyLen) == 0) {
(void)memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
(void)strncpy(childTableName, tag->value,
(tag->length < TSDB_TABLE_NAME_LEN ? tag->length : TSDB_TABLE_NAME_LEN));
tstrncpy(childTableName, tag->value, TMIN(TSDB_TABLE_NAME_LEN, tag->length + 1));
if (tsSmlDot2Underline) {
smlStrReplace(childTableName, strlen(childTableName));
}

10
source/client/src/clientStmt.c
View File

@ -236,7 +236,7 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags,
}
(void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
(void)strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
tstrncpy(pStmt->bInfo.tbFName, tbFName, TSDB_TABLE_FNAME_LEN);
pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
@ -1018,13 +1018,13 @@ int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
STMT_ERR_RET(stmtGetFromCache(pStmt));
if (pStmt->bInfo.needParse) {
(void)strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
STMT_ERR_RET(stmtParseSql(pStmt));
}
} else {
(void)strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
pStmt->exec.pRequest->requestId++;
pStmt->bInfo.needParse = false;
@ -1172,7 +1172,7 @@ int32_t stmtAppendTablePostHandle(STscStmt* pStmt, SStmtQNode* param) {
}
if (0 == pStmt->sql.siInfo.firstName[0]) {
(void)strcpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName);
tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
}
param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
@ -1313,7 +1313,7 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
// param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
param->restoreTbCols = false;
(void)strcpy(param->tblData.tbName, pStmt->bInfo.tbName);
tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
}
int64_t startUs3 = taosGetTimestampUs();

10
source/client/src/clientStmt2.c
View File

@ -187,7 +187,7 @@ static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void
}
(void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
(void)strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
tstrncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName));
pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
@ -961,13 +961,13 @@ int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
STMT_ERR_RET(stmtGetFromCache(pStmt));
if (pStmt->bInfo.needParse) {
(void)strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
STMT_ERR_RET(stmtParseSql(pStmt));
}
} else {
(void)strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
pStmt->exec.pRequest->requestId++;
pStmt->bInfo.needParse = false;
@ -1113,7 +1113,7 @@ static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
}
if (0 == pStmt->sql.siInfo.firstName[0]) {
(void)strcpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName);
tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
}
param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
@ -1367,7 +1367,7 @@ int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx) {
// param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
param->restoreTbCols = false;
(void)strcpy(param->tblData.tbName, pStmt->bInfo.tbName);
tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
}
int64_t startUs3 = taosGetTimestampUs();

10
source/client/src/clientTmq.c
View File

@ -993,7 +993,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
if (data == NULL) {
continue;
}
(void)strcpy(data->topicName, pTopic->topicName);
tstrncpy(data->topicName, pTopic->topicName, TSDB_TOPIC_FNAME_LEN);
data->offsetRows = taosArrayInit(numOfVgroups, sizeof(OffsetRows));
if (data->offsetRows == NULL) {
continue;
@ -1126,7 +1126,7 @@ static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopic
if (pVgEp == NULL) {
continue;
}
(void)sprintf(vgKey, "%s:%d", pTopic->topicName, pVgEp->vgId);
(void)snprintf(vgKey, sizeof(vgKey), "%s:%d", pTopic->topicName, pVgEp->vgId);
SVgroupSaveInfo* pInfo = taosHashGet(pVgOffsetHashMap, vgKey, strlen(vgKey));
STqOffsetVal offsetNew = {0};
@ -1187,7 +1187,7 @@ static void buildNewTopicList(tmq_t* tmq, SArray* newTopics, const SMqAskEpRsp*
continue;
}
char vgKey[TSDB_TOPIC_FNAME_LEN + 22] = {0};
(void)sprintf(vgKey, "%s:%d", pTopicCur->topicName, pVgCur->vgId);
(void)snprintf(vgKey, sizeof(vgKey), "%s:%d", pTopicCur->topicName, pVgCur->vgId);
char buf[TSDB_OFFSET_LEN] = {0};
tFormatOffset(buf, TSDB_OFFSET_LEN, &pVgCur->offsetInfo.endOffset);
@ -1993,7 +1993,7 @@ END:
if (pRspWrapper) {
pRspWrapper->code = code;
pRspWrapper->pollRsp.vgId = vgId;
(void)strcpy(pRspWrapper->pollRsp.topicName, pParam->topicName);
tstrncpy(pRspWrapper->pollRsp.topicName, pParam->topicName, TSDB_TOPIC_FNAME_LEN);
code = taosWriteQitem(tmq->mqueue, pRspWrapper);
if (code != 0) {
tmqFreeRspWrapper(pRspWrapper);
@ -2157,7 +2157,7 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p
}
pParam->refId = pTmq->refId;
(void)strcpy(pParam->topicName, pTopic->topicName);
tstrncpy(pParam->topicName, pTopic->topicName, TSDB_TOPIC_FNAME_LEN);
pParam->vgId = pVg->vgId;
pParam->requestId = req.reqId;

152
source/common/src/tdatablock.c
View File

@ -2446,9 +2446,11 @@ _error:
return NULL;
}
static char* formatTimestamp(char* buf, int32_t bufSize, int64_t val, int precision) {
static int32_t formatTimestamp(char* buf, size_t cap, int64_t val, int precision) {
time_t tt;
int32_t ms = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (precision == TSDB_TIME_PRECISION_NANO) {
tt = (time_t)(val / 1000000000);
ms = val % 1000000000;
@ -2460,14 +2462,6 @@ static char* formatTimestamp(char* buf, int32_t bufSize, int64_t val, int precis
ms = val % 1000;
}
/* comment out as it make testcases like select_with_tags.sim fail.
but in windows, this may cause the call to localtime crash if tt < 0,
need to find a better solution.
if (tt < 0) {
tt = 0;
}
*/
if (tt <= 0 && ms < 0) {
tt--;
if (precision == TSDB_TIME_PRECISION_NANO) {
@ -2479,24 +2473,40 @@ static char* formatTimestamp(char* buf, int32_t bufSize, int64_t val, int precis
}
}
struct tm ptm = {0};
if (taosLocalTime(&tt, &ptm, buf, bufSize) == NULL) {
return buf;
if (taosLocalTime(&tt, &ptm, buf, cap) == NULL) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
size_t pos = strftime(buf, bufSize, "%Y-%m-%d %H:%M:%S", &ptm);
size_t pos = strftime(buf, cap, "%Y-%m-%d %H:%M:%S", &ptm);
if (pos == 0) {
code = TSDB_CODE_OUT_OF_BUFFER;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t nwritten = 0;
if (precision == TSDB_TIME_PRECISION_NANO) {
sprintf(buf + pos, ".%09d", ms);
nwritten = snprintf(buf + pos, cap - pos, ".%09d", ms);
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
sprintf(buf + pos, ".%06d", ms);
nwritten = snprintf(buf + pos, cap - pos, ".%06d", ms);
} else {
sprintf(buf + pos, ".%03d", ms);
nwritten = snprintf(buf + pos, cap - pos, ".%03d", ms);
}
return buf;
if (nwritten >= cap - pos) {
code = TSDB_CODE_OUT_OF_BUFFER;
TSDB_CHECK_CODE(code, lino, _end);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
// for debug
int32_t dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf, const char* taskIdStr) {
int32_t lino = 0;
int32_t size = 2048 * 1024;
int32_t code = 0;
char* dumpBuf = NULL;
@ -2530,6 +2540,7 @@ int32_t dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
if (pColInfoData == NULL) {
code = terrno;
lino = __LINE__;
goto _exit;
}
@ -2543,7 +2554,10 @@ int32_t dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP:
memset(pBuf, 0, sizeof(pBuf));
(void)formatTimestamp(pBuf, sizeof(pBuf), *(uint64_t*)var, pColInfoData->info.precision);
code = formatTimestamp(pBuf, sizeof(pBuf), *(uint64_t*)var, pColInfoData->info.precision);
if (code != TSDB_CODE_SUCCESS) {
snprintf(pBuf, sizeof(pBuf), "NaN");
}
len += snprintf(dumpBuf + len, size - len, " %25s |", pBuf);
if (len >= size - 1) goto _exit;
break;
@ -2609,6 +2623,7 @@ int32_t dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf
code = taosUcs4ToMbs((TdUcs4*)varDataVal(pData), dataSize, pBuf);
if (code < 0) {
uError("func %s failed to convert to ucs charset since %s", __func__, tstrerror(code));
lino = __LINE__;
goto _exit;
} else { // reset the length value
code = TSDB_CODE_SUCCESS;
@ -2629,7 +2644,7 @@ _exit:
*pDataBuf = dumpBuf;
dumpBuf = NULL;
} else {
uError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
if (dumpBuf) {
taosMemoryFree(dumpBuf);
}
@ -2860,27 +2875,98 @@ _end:
return code;
}
void buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId) {
char tmp[TSDB_TABLE_NAME_LEN] = {0};
if (stbName == NULL){
snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%"PRIu64, groupId);
}else{
// Construct the child table name in the form of <ctbName>_<stbName>_<groupId> and store it in `ctbName`.
// If the name length exceeds TSDB_TABLE_NAME_LEN, first convert <stbName>_<groupId> to an MD5 value and then
// concatenate. If the length is still too long, convert <ctbName> to an MD5 value as well.
int32_t buildCtbNameAddGroupId(const char* stbName, char* ctbName, uint64_t groupId, size_t cap) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
char tmp[TSDB_TABLE_NAME_LEN] = {0};
char* suffix = tmp;
size_t suffixCap = sizeof(tmp);
size_t suffixLen = 0;
size_t prefixLen = 0;
T_MD5_CTX context;
if (ctbName == NULL || cap < TSDB_TABLE_NAME_LEN) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
prefixLen = strlen(ctbName);
if (stbName == NULL) {
suffixLen = snprintf(suffix, suffixCap, "%" PRIu64, groupId);
if (suffixLen >= suffixCap) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
} else {
int32_t i = strlen(stbName) - 1;
for(; i >= 0; i--){
if (stbName[i] == '.'){
for (; i >= 0; i--) {
if (stbName[i] == '.') {
break;
}
}
snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%s_%"PRIu64, stbName + i + 1, groupId);
}
ctbName[TSDB_TABLE_NAME_LEN - strlen(tmp) - 1] = 0; // put stbname + groupId to the end
(void)strcat(ctbName, tmp);
for(int i = 0; i < strlen(ctbName); i++){
if(ctbName[i] == '.'){
ctbName[i] = '_';
suffixLen = snprintf(suffix, suffixCap, "%s_%" PRIu64, stbName + i + 1, groupId);
if (suffixLen >= suffixCap) {
suffixCap = suffixLen + 1;
suffix = taosMemoryMalloc(suffixCap);
TSDB_CHECK_NULL(suffix, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
suffixLen = snprintf(suffix, suffixCap, "%s_%" PRIu64, stbName + i + 1, groupId);
if (suffixLen >= suffixCap) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
}
}
if (prefixLen + suffixLen + 1 >= TSDB_TABLE_NAME_LEN) {
// If the name length exceeeds the limit, convert the suffix to MD5 value.
tMD5Init(&context);
tMD5Update(&context, (uint8_t*)suffix, suffixLen);
tMD5Final(&context);
suffixLen = snprintf(suffix, suffixCap, "%016" PRIx64 "%016" PRIx64, *(uint64_t*)context.digest,
*(uint64_t*)(context.digest + 8));
if (suffixLen >= suffixCap) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
}
if (prefixLen + suffixLen + 1 >= TSDB_TABLE_NAME_LEN) {
// If the name is still too long, convert the ctbName to MD5 value.
tMD5Init(&context);
tMD5Update(&context, (uint8_t*)ctbName, prefixLen);
tMD5Final(&context);
prefixLen = snprintf(ctbName, cap, "t_%016" PRIx64 "%016" PRIx64, *(uint64_t*)context.digest,
*(uint64_t*)(context.digest + 8));
if (prefixLen >= cap) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
}
if (prefixLen + suffixLen + 1 >= TSDB_TABLE_NAME_LEN) {
code = TSDB_CODE_INTERNAL_ERROR;
TSDB_CHECK_CODE(code, lino, _end);
}
ctbName[prefixLen] = '_';
tstrncpy(&ctbName[prefixLen + 1], suffix, cap - prefixLen - 1);
for (char* p = ctbName; *p; ++p) {
if (*p == '.') *p = '_';
}
_end:
if (code != TSDB_CODE_SUCCESS) {
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
if (suffix != tmp) {
taosMemoryFree(suffix);
}
return code;
}
// auto stream subtable name starts with 't_', followed by the first segment of MD5 digest for group vals.

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

@ -1725,6 +1725,7 @@ int32_t taosReadDataFolder(const char *cfgDir, const char **envCmd, const char *
TAOS_CHECK_RETURN(cfgInit(&pCfg));
TAOS_CHECK_GOTO(cfgAddDir(pCfg, "dataDir", tsDataDir, CFG_SCOPE_SERVER, CFG_DYN_NONE), NULL, _exit);
TAOS_CHECK_GOTO(cfgAddInt32(pCfg, "debugFlag", dDebugFlag, 0, 255, CFG_SCOPE_SERVER, CFG_DYN_SERVER), NULL, _exit);
TAOS_CHECK_GOTO(cfgAddInt32(pCfg, "dDebugFlag", dDebugFlag, 0, 255, CFG_SCOPE_SERVER, CFG_DYN_SERVER) ,NULL, _exit);
if ((code = taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl)) != 0) {

56
source/common/test/dataformatTest.cpp
View File

@ -480,7 +480,7 @@ TEST(testCase, StreamAllNormTest) {
char ctbName[TSDB_TABLE_NAME_LEN] = {0};
uint64_t groupId = 12345;
buildCtbNameAddGroupId(NULL, ctbName, groupId);
buildCtbNameAddGroupId(NULL, ctbName, groupId, sizeof(ctbName));
ASSERT_STREQ("_12345", ctbName);
}
@ -490,7 +490,7 @@ TEST(testCase, StreamWithStbName) {
char ctbName[TSDB_TABLE_NAME_LEN] = {0};
uint64_t groupId = 12345;
buildCtbNameAddGroupId(stbName, ctbName, groupId);
buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName));
ASSERT_STREQ("_stb_12345", ctbName);
}
@ -500,7 +500,7 @@ TEST(testCase, StreamWithoutDotInStbName) {
char ctbName[TSDB_TABLE_NAME_LEN] = {0};
uint64_t groupId = 12345;
buildCtbNameAddGroupId(stbName, ctbName, groupId);
buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName));
ASSERT_STREQ("_table_12345", ctbName);
}
@ -510,11 +510,59 @@ TEST(testCase, StreamWithoutDotInStbName2) {
char ctbName[TSDB_TABLE_NAME_LEN] = {0};
uint64_t groupId = 12345;
buildCtbNameAddGroupId(stbName, ctbName, groupId);
buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName));
ASSERT_STREQ("__12345", ctbName);
}
TEST(testCase, StreamWithLongStbName) {
char ctbName[TSDB_TABLE_NAME_LEN];
char expectName[TSDB_TABLE_NAME_LEN];
char *stbName = "a_simle_stb_name";
uint64_t groupId = UINT64_MAX;
// test basic function
strcpy(ctbName, "a_simple_ctb_name");
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName)), TSDB_CODE_SUCCESS);
EXPECT_STREQ(ctbName, "a_simple_ctb_name_a_simle_stb_name_18446744073709551615");
// test null stbName
strcpy(ctbName, "a_simple_ctb_name");
stbName = NULL;
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName)), TSDB_CODE_SUCCESS);
EXPECT_STREQ(ctbName, "a_simple_ctb_name_18446744073709551615");
// test buffer capcity check
EXPECT_EQ(buildCtbNameAddGroupId(stbName, NULL, groupId, sizeof(ctbName)), TSDB_CODE_INTERNAL_ERROR);
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName) - 1), TSDB_CODE_INTERNAL_ERROR);
// test md5 conversion of stbName with groupid
for (int32_t i = 0; i < 159; ++i) ctbName[i] = 'A';
ctbName[159] = '\0';
stbName = taosStrdup(ctbName);
snprintf(expectName, TSDB_TABLE_NAME_LEN, "%s_d85f0d87946d76eeedd7b7b78b7492a2", ctbName);
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName)), TSDB_CODE_SUCCESS);
EXPECT_STREQ(ctbName, expectName);
// test md5 conversion of all parts
for (int32_t i = 0; i < 190; ++i) ctbName[i] = 'A';
ctbName[190] = '\0';
tstrncpy(expectName, "t_d38a8b2df999bef0082ffc80a59a9cd7_d85f0d87946d76eeedd7b7b78b7492a2", TSDB_TABLE_NAME_LEN);
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName)), TSDB_CODE_SUCCESS);
EXPECT_STREQ(ctbName, expectName);
// test larger stbName
taosMemoryFree(stbName);
for (int32_t i = 0; i < 190; ++i) ctbName[i] = 'A';
ctbName[190] = '\0';
stbName = taosStrdup(ctbName);
tstrncpy(expectName, "t_d38a8b2df999bef0082ffc80a59a9cd7_9c99cc7c52073b63fb750af402d9b84b", TSDB_TABLE_NAME_LEN);
EXPECT_EQ(buildCtbNameAddGroupId(stbName, ctbName, groupId, sizeof(ctbName)), TSDB_CODE_SUCCESS);
EXPECT_STREQ(ctbName, expectName);
taosMemoryFree(stbName);
}
#if 1
TEST(testCase, NoneTest) {
const static int nCols = 14;

23
source/dnode/mnode/impl/src/mndAnode.c
View File

@ -43,7 +43,7 @@ static void mndCancelGetNextAnode(SMnode *pMnode, void *pIter);
static int32_t mndRetrieveAnodesFull(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextAnodeFull(SMnode *pMnode, void *pIter);
static int32_t mndGetAnodeAlgoList(const char *url, SAnodeObj *pObj);
static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status);
static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status, int32_t statusLen);
int32_t mndInitAnode(SMnode *pMnode) {
SSdbTable table = {
@ -603,7 +603,7 @@ static int32_t mndRetrieveAnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
if (code != 0) goto _end;
status[0] = 0;
if (mndGetAnodeStatus(pObj, status) == 0) {
if (mndGetAnodeStatus(pObj, status, 64) == 0) {
STR_TO_VARSTR(buf, status);
} else {
STR_TO_VARSTR(buf, "offline");
@ -698,7 +698,7 @@ static int32_t mndDecodeAlgoList(SJson *pJson, SAnodeObj *pObj) {
code = tjsonGetDoubleValue(pJson, "protocol", &tmp);
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
protocol = (int32_t)(tmp * 1000);
if (protocol != 100) return TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;
if (protocol != 100 && protocol != 1000) return TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;
code = tjsonGetDoubleValue(pJson, "version", &tmp);
pObj->version = (int32_t)(tmp * 1000);
@ -722,10 +722,10 @@ static int32_t mndDecodeAlgoList(SJson *pJson, SAnodeObj *pObj) {
SJson *detail = tjsonGetArrayItem(details, d);
if (detail == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
code = tjsonGetStringValue(detail, "type", buf);
code = tjsonGetStringValue2(detail, "type", buf, sizeof(buf));
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
EAnalAlgoType type = taosAnalAlgoInt(buf);
if (type < 0 || type >= ANAL_ALGO_TYPE_END) continue;
if (type < 0 || type >= ANAL_ALGO_TYPE_END) return TSDB_CODE_MND_ANODE_INVALID_ALGO_TYPE;
SJson *algos = tjsonGetObjectItem(detail, "algo");
if (algos == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
@ -734,13 +734,12 @@ static int32_t mndDecodeAlgoList(SJson *pJson, SAnodeObj *pObj) {
SJson *algo = tjsonGetArrayItem(algos, a);
if (algo == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;
code = tjsonGetStringValue(algo, "name", buf);
if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;
code = tjsonGetStringValue2(algo, "name", buf, sizeof(buf));
if (code < 0) return TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME;
SAnodeAlgo algoObj = {0};
algoObj.nameLen = strlen(buf) + 1;
if (algoObj.nameLen > TSDB_ANAL_ALGO_NAME_LEN) return TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME;
if (algoObj.nameLen <= 1) return TSDB_CODE_OUT_OF_MEMORY;
if (algoObj.nameLen <= 1) return TSDB_CODE_INVALID_JSON_FORMAT;
algoObj.name = taosMemoryCalloc(algoObj.nameLen, 1);
tstrncpy(algoObj.name, buf, algoObj.nameLen);
@ -764,7 +763,7 @@ static int32_t mndGetAnodeAlgoList(const char *url, SAnodeObj *pObj) {
TAOS_RETURN(code);
}
static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status) {
static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status, int32_t statusLen) {
int32_t code = 0;
int32_t protocol = 0;
double tmp = 0;
@ -780,12 +779,12 @@ static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status) {
goto _OVER;
}
protocol = (int32_t)(tmp * 1000);
if (protocol != 100) {
if (protocol != 100 && protocol != 1000) {
code = TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;
goto _OVER;
}
code = tjsonGetStringValue(pJson, "status", status);
code = tjsonGetStringValue2(pJson, "status", status, statusLen);
if (code < 0) {
code = TSDB_CODE_INVALID_JSON_FORMAT;
goto _OVER;

8
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -167,7 +167,7 @@ static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRs
}
STopicPrivilege *data = taosArrayReserve(rsp->topicPrivileges, 1);
MND_TMQ_NULL_CHECK(data);
(void)strcpy(data->topic, topic);
tstrncpy(data->topic, topic, TSDB_TOPIC_FNAME_LEN);
if (mndCheckTopicPrivilege(pMnode, user, MND_OPER_SUBSCRIBE, pTopic) != 0 ||
grantCheckExpire(TSDB_GRANT_SUBSCRIPTION) < 0) {
data->noPrivilege = 1;
@ -278,7 +278,7 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
taosRLockLatch(&pSub->lock);
SMqSubTopicEp topicEp = {0};
(void)strcpy(topicEp.topic, topic);
tstrncpy(topicEp.topic, topic, TSDB_TOPIC_FNAME_LEN);
// 2.1 fetch topic schema
SMqTopicObj *pTopic = NULL;
@ -910,7 +910,7 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
// consumer id
char consumerIdHex[TSDB_CONSUMER_ID_LEN + VARSTR_HEADER_SIZE] = {0};
(void)sprintf(varDataVal(consumerIdHex), "0x%" PRIx64, pConsumer->consumerId);
(void)snprintf(varDataVal(consumerIdHex), TSDB_CONSUMER_ID_LEN, "0x%" PRIx64, pConsumer->consumerId);
varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
@ -993,7 +993,7 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
parasStr = taosMemoryCalloc(1, pShow->pMeta->pSchemas[cols].bytes);
MND_TMQ_NULL_CHECK(parasStr);
(void)sprintf(varDataVal(parasStr), "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName,
(void)snprintf(varDataVal(parasStr), pShow->pMeta->pSchemas[cols].bytes - VARSTR_HEADER_SIZE, "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName,
pConsumer->autoCommit, pConsumer->autoCommitInterval, buf);
varDataSetLen(parasStr, strlen(varDataVal(parasStr)));

2
source/dnode/mnode/impl/src/mndMain.c
View File

@ -575,6 +575,8 @@ static int32_t mndOpenSdb(SMnode *pMnode) {
code = sdbReadFile(pMnode->pSdb);
}
mInfo("vgId:1, mnode sdb is opened, with applied index:%" PRId64, pMnode->pSdb->commitIndex);
atomic_store_64(&pMnode->applied, pMnode->pSdb->commitIndex);
return code;
}

9
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -187,12 +187,12 @@ static void mndSplitSubscribeKey(const char *key, char *topic, char *cgroup, boo
(void)memcpy(cgroup, key, i);
cgroup[i] = 0;
if (fullName) {
(void)strcpy(topic, &key[i + 1]);
tstrncpy(topic, &key[i + 1], TSDB_TOPIC_FNAME_LEN);
} else {
while (key[i] != '.') {
i++;
}
(void)strcpy(topic, &key[i + 1]);
tstrncpy(topic, &key[i + 1], TSDB_CGROUP_LEN);
}
}
@ -1361,7 +1361,7 @@ static int32_t buildResult(SSDataBlock *pBlock, int32_t *numOfRows, int64_t cons
// consumer id
char consumerIdHex[TSDB_CONSUMER_ID_LEN] = {0};
(void)sprintf(varDataVal(consumerIdHex), "0x%" PRIx64, consumerId);
(void)snprintf(varDataVal(consumerIdHex), TSDB_CONSUMER_ID_LEN - VARSTR_HEADER_SIZE, "0x%" PRIx64, consumerId);
varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
@ -1398,7 +1398,8 @@ static int32_t buildResult(SSDataBlock *pBlock, int32_t *numOfRows, int64_t cons
// vg id
char buf[TSDB_OFFSET_LEN * 2 + VARSTR_HEADER_SIZE] = {0};
(void)tFormatOffset(varDataVal(buf), TSDB_OFFSET_LEN, &data->offset);
(void)sprintf(varDataVal(buf) + strlen(varDataVal(buf)), "/%" PRId64, data->ever);
(void)snprintf(varDataVal(buf) + strlen(varDataVal(buf)),
sizeof(buf) - VARSTR_HEADER_SIZE - strlen(varDataVal(buf)), "/%" PRId64, data->ever);
varDataSetLen(buf, strlen(varDataVal(buf)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
MND_TMQ_NULL_CHECK(pColInfo);

2
source/dnode/mnode/impl/src/mndSync.c
View File

@ -637,7 +637,7 @@ void mndSyncStop(SMnode *pMnode) {
(void)taosThreadMutexLock(&pMgmt->lock);
if (pMgmt->transId != 0) {
mInfo("vgId:1, is stopped and post sem, trans:%d", pMgmt->transId);
mInfo("vgId:1, trans:%d, is stopped and post sem", pMgmt->transId);
pMgmt->transId = 0;
pMgmt->transSec = 0;
pMgmt->errCode = TSDB_CODE_APP_IS_STOPPING;

13
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -589,6 +589,7 @@ STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
SSdb *pSdb = pMnode->pSdb;
if (pTrans != NULL) mInfo("vgId:1, trans:%d, release transaction", pTrans->id);
sdbRelease(pSdb, pTrans);
}
@ -1131,10 +1132,11 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
if (!sendRsp) {
return;
} else {
mInfo("trans:%d, send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id, mndTransStr(pTrans->stage),
pTrans->failedTimes, code);
mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
mndTransStr(pTrans->stage), pTrans->failedTimes, code);
}
mInfo("vgId:1, trans:%d, start to lock rpc array", pTrans->id);
taosWLockLatch(&pTrans->lockRpcArray);
int32_t size = taosArrayGetSize(pTrans->pRpcArray);
if (size <= 0) {
@ -1155,8 +1157,8 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
if (i != 0 && code == 0) {
code = TSDB_CODE_MNODE_NOT_FOUND;
}
mInfo("trans:%d, client:%d send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code, mndTransStr(pTrans->stage),
pInfo->ahandle);
mInfo("vgId:1, trans:%d, client:%d start to send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code,
mndTransStr(pTrans->stage), pInfo->ahandle);
SRpcMsg rspMsg = {.code = code, .info = *pInfo};
@ -1199,6 +1201,9 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
}
tmsgSendRsp(&rspMsg);
mInfo("vgId:1, trans:%d, client:%d send rsp finished, code:0x%x stage:%s app:%p", pTrans->id, i, code,
mndTransStr(pTrans->stage), pInfo->ahandle);
}
}
taosArrayClear(pTrans->pRpcArray);

9
source/dnode/mnode/sdb/src/sdb.c
View File

@ -168,11 +168,10 @@ static int32_t sdbCreateDir(SSdb *pSdb) {
}
void sdbSetApplyInfo(SSdb *pSdb, int64_t index, int64_t term, int64_t config) {
#if 1
mTrace("mnode apply info changed from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " to index:%" PRId64
" term:%" PRId64 " config:%" PRId64,
pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, config);
#endif
mInfo("vgId:1, mnode apply info changed from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " to index:%" PRId64
" term:%" PRId64 " config:%" PRId64,
pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, config);
pSdb->applyIndex = index;
pSdb->applyTerm = term;
pSdb->applyConfig = config;

5
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -207,6 +207,8 @@ static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) {
return terrno;
}
mInfo("vgId:1, write sdb file with sdb applyIndex:%" PRId64 " term:%" PRId64 " config:%" PRId64, pSdb->applyIndex,
pSdb->applyTerm, pSdb->applyConfig);
if (taosWriteFile(pFile, &pSdb->applyIndex, sizeof(int64_t)) != sizeof(int64_t)) {
return terrno;
}
@ -607,6 +609,9 @@ int32_t sdbWriteFile(SSdb *pSdb, int32_t delta) {
}
if (code != 0) {
mError("failed to write sdb file since %s", tstrerror(code));
} else {
mInfo("write sdb file success, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64, pSdb->applyIndex,
pSdb->applyTerm, pSdb->applyConfig);
}
(void)taosThreadMutexUnlock(&pSdb->filelock);
return code;

1
source/dnode/mnode/sdb/src/sdbRow.c
View File

@ -42,6 +42,7 @@ void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc) {
// remove attached object such as trans
SdbDeleteFp deleteFp = pSdb->deleteFps[pRow->type];
if (deleteFp != NULL) {
mInfo("vgId:1, deleteFp:%p, type:%s", deleteFp, sdbTableName(pRow->type));
(void)(*deleteFp)(pSdb, pRow->pObj, callFunc);
}

7
source/dnode/vnode/src/meta/metaTable.c
View File

@ -3133,7 +3133,12 @@ static void colCompressDebug(SHashObj *pColCmprObj) {
int32_t metaGetColCmpr(SMeta *pMeta, tb_uid_t uid, SHashObj **ppColCmprObj) {
int rc = 0;
SHashObj *pColCmprObj = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
SHashObj *pColCmprObj = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
if (pColCmprObj == NULL) {
pColCmprObj = NULL;
return TSDB_CODE_OUT_OF_MEMORY;
}
void *pData = NULL;
int nData = 0;
SMetaEntry e = {0};

25
source/dnode/vnode/src/tq/tqSink.c
View File

@ -73,14 +73,19 @@ int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* p
}
if (varTbName != NULL && varTbName != (void*)-1) {
name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
size_t cap = TMAX(TSDB_TABLE_NAME_LEN, varDataLen(varTbName) + 1);
name = taosMemoryMalloc(cap);
if (name == NULL) {
return terrno;
}
memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
name[varDataLen(varTbName)] = '\0';
if (newSubTableRule && !isAutoTableName(name) && !alreadyAddGroupId(name, groupId) && groupId != 0 && stbFullName) {
buildCtbNameAddGroupId(stbFullName, name, groupId);
int32_t code = buildCtbNameAddGroupId(stbFullName, name, groupId, cap);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
} else if (stbFullName) {
int32_t code = buildCtbNameByGroupId(stbFullName, groupId, &name);
@ -107,7 +112,7 @@ int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* p
groupId, name, skey, ekey);
SSingleDeleteReq req = {.startTs = skey, .endTs = ekey};
strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN - 1);
tstrncpy(req.tbname, name, TSDB_TABLE_NAME_LEN);
void* p = taosArrayPush(deleteReq->deleteReqs, &req);
if (p == NULL) {
return terrno;
@ -235,8 +240,11 @@ int32_t setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock*
return terrno;
}
strcpy(pCreateTableReq->name, pDataBlock->info.parTbName);
buildCtbNameAddGroupId(stbFullName, pCreateTableReq->name, gid);
tstrncpy(pCreateTableReq->name, pDataBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
int32_t code = buildCtbNameAddGroupId(stbFullName, pCreateTableReq->name, gid, TSDB_TABLE_NAME_LEN);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
// tqDebug("gen name from:%s", pDataBlock->info.parTbName);
} else {
pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
@ -852,9 +860,12 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
!alreadyAddGroupId(dstTableName, groupId) && groupId != 0) {
tqDebug("s-task:%s append groupId:%" PRId64 " for generated dstTable:%s", id, groupId, dstTableName);
if (pTask->ver == SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
buildCtbNameAddGroupId(NULL, dstTableName, groupId);
code = buildCtbNameAddGroupId(NULL, dstTableName, groupId, sizeof(pDataBlock->info.parTbName));
} else if (pTask->ver > SSTREAM_TASK_SUBTABLE_CHANGED_VER && stbFullName) {
buildCtbNameAddGroupId(stbFullName, dstTableName, groupId);
code = buildCtbNameAddGroupId(stbFullName, dstTableName, groupId, sizeof(pDataBlock->info.parTbName));
}
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
}

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

@ -1767,10 +1767,14 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
SLastCol *pLastCol = h ? (SLastCol *)taosLRUCacheValue(pCache, h) : NULL;
if (h && pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) {
SLastCol lastCol = *pLastCol;
TAOS_CHECK_GOTO(tsdbCacheReallocSLastCol(&lastCol, NULL), NULL, _exit);
if (TSDB_CODE_SUCCESS != (code = tsdbCacheReallocSLastCol(&lastCol, NULL))) {
tsdbLRUCacheRelease(pCache, h, false);
TAOS_CHECK_GOTO(code, NULL, _exit);
}
if (taosArrayPush(pLastArray, &lastCol) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
} else {
@ -1780,28 +1784,33 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
if (taosArrayPush(pLastArray, &noneCol) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
if (!remainCols) {
if ((remainCols = taosArrayInit(numKeys, sizeof(SIdxKey))) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
}
if (!ignoreFromRocks) {
if ((ignoreFromRocks = taosArrayInit(numKeys, sizeof(bool))) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
}
if (taosArrayPush(remainCols, &(SIdxKey){i, key}) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
bool ignoreRocks = pLastCol ? (pLastCol->cacheStatus == TSDB_LAST_CACHE_NO_CACHE) : false;
if (taosArrayPush(ignoreFromRocks, &ignoreRocks) == NULL) {
code = terrno;
tsdbLRUCacheRelease(pCache, h, false);
goto _exit;
}
}
@ -1822,6 +1831,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
SLastCol lastCol = *pLastCol;
code = tsdbCacheReallocSLastCol(&lastCol, NULL);
if (code) {
tsdbLRUCacheRelease(pCache, h, false);
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
}

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

@ -433,6 +433,21 @@ static int32_t tsdbFSCreateFileObjHash(STFileSystem *fs, STFileHash *hash) {
if (fset->farr[i] != NULL) {
code = tsdbFSAddEntryToFileObjHash(hash, fset->farr[i]->fname);
TSDB_CHECK_CODE(code, lino, _exit);
if (TSDB_FTYPE_DATA == i && fset->farr[i]->f->lcn > 0) {
STFileObj *fobj = fset->farr[i];
int32_t lcn = fobj->f->lcn;
char lcn_name[TSDB_FILENAME_LEN];
snprintf(lcn_name, TSDB_FQDN_LEN, "%s", fobj->fname);
char *dot = strrchr(lcn_name, '.');
if (dot) {
snprintf(dot + 1, TSDB_FQDN_LEN - (dot + 1 - lcn_name), "%d.data", lcn);
code = tsdbFSAddEntryToFileObjHash(hash, lcn_name);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
}
}
@ -535,9 +550,7 @@ static int32_t tsdbFSDoSanAndFix(STFileSystem *fs) {
for (const STfsFile *file = NULL; (file = tfsReaddir(dir)) != NULL;) {
if (taosIsDir(file->aname)) continue;
if (tsdbFSGetFileObjHashEntry(&fobjHash, file->aname) == NULL &&
strncmp(file->aname + strlen(file->aname) - 3, ".cp", 3) &&
strncmp(file->aname + strlen(file->aname) - 5, ".data", 5)) {
if (tsdbFSGetFileObjHashEntry(&fobjHash, file->aname) == NULL) {
tsdbRemoveFile(file->aname);
}
}

7
source/dnode/vnode/src/tsdb/tsdbFSetRW.c
View File

@ -46,8 +46,11 @@ static int32_t tsdbFSetWriteTableDataBegin(SFSetWriter *writer, const TABLEID *t
code = tsdbUpdateSkmTb(writer->config->tsdb, writer->ctx->tbid, writer->skmTb);
TSDB_CHECK_CODE(code, lino, _exit);
if (writer->pColCmprObj != NULL) {
taosHashCleanup(writer->pColCmprObj);
writer->pColCmprObj = NULL;
}
code = metaGetColCmpr(writer->config->tsdb->pVnode->pMeta, tbid->suid ? tbid->suid : tbid->uid, &writer->pColCmprObj);
// TODO: TSDB_CHECK_CODE(code, lino, _exit);
writer->blockDataIdx = 0;
for (int32_t i = 0; i < ARRAY_SIZE(writer->blockData); i++) {
@ -127,6 +130,8 @@ _exit:
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
taosHashCleanup(writer->pColCmprObj);
writer->pColCmprObj = NULL;
return code;
}

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

@ -1635,13 +1635,17 @@ int32_t getTableListInfo(const SExecTaskInfo* pTaskInfo, SArray** pList) {
int32_t qStreamOperatorReleaseState(qTaskInfo_t tInfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;
pTaskInfo->pRoot->fpSet.releaseStreamStateFn(pTaskInfo->pRoot);
if (pTaskInfo->pRoot->fpSet.releaseStreamStateFn != NULL) {
pTaskInfo->pRoot->fpSet.releaseStreamStateFn(pTaskInfo->pRoot);
}
return 0;
}
int32_t qStreamOperatorReloadState(qTaskInfo_t tInfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;
pTaskInfo->pRoot->fpSet.reloadStreamStateFn(pTaskInfo->pRoot);
if (pTaskInfo->pRoot->fpSet.reloadStreamStateFn != NULL) {
pTaskInfo->pRoot->fpSet.reloadStreamStateFn(pTaskInfo->pRoot);
}
return 0;
}

81
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -49,7 +49,7 @@
#define STREAM_SESSION_OP_CHECKPOINT_NAME "StreamSessionOperator_Checkpoint"
#define STREAM_STATE_OP_CHECKPOINT_NAME "StreamStateOperator_Checkpoint"
#define MAX_STREAM_HISTORY_RESULT 20000000
#define MAX_STREAM_HISTORY_RESULT 20000000
typedef struct SStateWindowInfo {
SResultWindowInfo winInfo;
@ -449,7 +449,7 @@ void destroyFlusedPos(void* pRes) {
}
void destroyFlusedppPos(void* ppRes) {
void *pRes = *(void **)ppRes;
void* pRes = *(void**)ppRes;
destroyFlusedPos(pRes);
}
@ -507,7 +507,7 @@ void destroyStreamFinalIntervalOperatorInfo(void* param) {
}
taosArrayDestroy(pInfo->pMidPullDatas);
if (pInfo->pState !=NULL && pInfo->pState->dump == 1) {
if (pInfo->pState != NULL && pInfo->pState->dump == 1) {
taosMemoryFreeClear(pInfo->pState->pTdbState->pOwner);
taosMemoryFreeClear(pInfo->pState->pTdbState);
}
@ -548,7 +548,8 @@ void reloadFromDownStream(SOperatorInfo* downstream, SStreamIntervalOperatorInfo
bool hasSrcPrimaryKeyCol(SSteamOpBasicInfo* pInfo) { return pInfo->primaryPkIndex != -1; }
int32_t initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SStreamIntervalOperatorInfo* pInfo, struct SSteamOpBasicInfo* pBasic) {
int32_t initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SStreamIntervalOperatorInfo* pInfo,
struct SSteamOpBasicInfo* pBasic) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStateStore* pAPI = &downstream->pTaskInfo->storageAPI.stateStore;
@ -1028,7 +1029,7 @@ static int32_t getNextQualifiedFinalWindow(SInterval* pInterval, STimeWindow* pN
}
static int32_t doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, uint64_t groupId,
SSHashObj* pUpdatedMap, SSHashObj* pDeletedMap) {
SSHashObj* pUpdatedMap, SSHashObj* pDeletedMap) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperator->info;
@ -1661,7 +1662,7 @@ static int32_t doStreamFinalIntervalAggNext(SOperatorInfo* pOperator, SSDataBloc
pInfo->binfo.pRes->info.type = pBlock->info.type;
} else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
pBlock->info.type == STREAM_CLEAR) {
SArray* delWins = taosArrayInit(8, sizeof(SWinKey));
SArray* delWins = taosArrayInit(8, sizeof(SWinKey));
QUERY_CHECK_NULL(delWins, code, lino, _end, terrno);
SHashObj* finalMap = IS_FINAL_INTERVAL_OP(pOperator) ? pInfo->pFinalPullDataMap : NULL;
code = doDeleteWindows(pOperator, &pInfo->interval, pBlock, delWins, pInfo->pUpdatedMap, finalMap);
@ -1897,9 +1898,8 @@ _end:
}
}
int32_t createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo, int32_t numOfChild,
SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {
int32_t createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo,
int32_t numOfChild, SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {
QRY_PARAM_CHECK(pOptrInfo);
int32_t code = TSDB_CODE_SUCCESS;
@ -1959,8 +1959,8 @@ int32_t createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiN
pAPI->stateStore.streamStateSetNumber(pInfo->pState, -1, pInfo->primaryTsIndex);
int32_t numOfCols = 0;
SExprInfo* pExprInfo = NULL;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = NULL;
code = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &pExprInfo, &numOfCols);
QUERY_CHECK_CODE(code, lino, _error);
@ -2042,11 +2042,13 @@ int32_t createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiN
pOperator->info = pInfo;
if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) {
pOperator->fpSet = createOperatorFpSet(NULL, doStreamMidIntervalAggNext, NULL, destroyStreamFinalIntervalOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
pOperator->fpSet =
createOperatorFpSet(NULL, doStreamMidIntervalAggNext, NULL, destroyStreamFinalIntervalOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
} else {
pOperator->fpSet = createOperatorFpSet(NULL, doStreamFinalIntervalAggNext, NULL, destroyStreamFinalIntervalOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
pOperator->fpSet =
createOperatorFpSet(NULL, doStreamFinalIntervalAggNext, NULL, destroyStreamFinalIntervalOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
}
setOperatorStreamStateFn(pOperator, streamIntervalReleaseState, streamIntervalReloadState);
if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||
@ -2221,9 +2223,9 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SExprSupp* pExpSup, in
pSup->stateStore.streamStateSetNumber(pSup->pState, -1, tsIndex);
int32_t funResSize = getMaxFunResSize(pExpSup, numOfOutput);
pSup->pState->pFileState = NULL;
code = pSup->stateStore.streamFileStateInit(
tsStreamBufferSize, sizeof(SSessionKey), pSup->resultRowSize, funResSize, sesionTs, pSup->pState,
pTwAggSup->deleteMark, taskIdStr, pHandle->checkpointId, STREAM_STATE_BUFF_SORT, &pSup->pState->pFileState);
code = pSup->stateStore.streamFileStateInit(tsStreamBufferSize, sizeof(SSessionKey), pSup->resultRowSize, funResSize,
sesionTs, pSup->pState, pTwAggSup->deleteMark, taskIdStr,
pHandle->checkpointId, STREAM_STATE_BUFF_SORT, &pSup->pState->pFileState);
QUERY_CHECK_CODE(code, lino, _end);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
@ -3309,8 +3311,8 @@ int32_t doStreamSessionDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpe
QUERY_CHECK_CONDITION((winCode == TSDB_CODE_SUCCESS), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
code =
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &winfo.sessionWin, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
code = tSimpleHashPut(pInfo->streamAggSup.pResultRows, &winfo.sessionWin, sizeof(SSessionKey), &winfo,
sizeof(SResultWindowInfo));
QUERY_CHECK_CODE(code, lino, _end);
}
@ -3772,8 +3774,8 @@ _end:
}
}
int32_t createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {
int32_t createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo,
SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {
QRY_PARAM_CHECK(pOptrInfo);
SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
@ -3807,7 +3809,7 @@ int32_t createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode
QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
pInfo->binfo.pRes = pResBlock;
SExprInfo* pExprInfo = NULL;
SExprInfo* pExprInfo = NULL;
code = createExprInfo(pSessionNode->window.pFuncs, NULL, &pExprInfo, &numOfCols);
QUERY_CHECK_CODE(code, lino, _error);
@ -3886,8 +3888,9 @@ int32_t createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode
QUERY_CHECK_CODE(code, lino, _error);
}
}
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamSessionAggNext, NULL, destroyStreamSessionAggOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
pOperator->fpSet =
createOperatorFpSet(optrDummyOpenFn, doStreamSessionAggNext, NULL, destroyStreamSessionAggOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamSessionReleaseState, streamSessionReloadState);
if (downstream) {
@ -4102,8 +4105,8 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
}
int32_t createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo, int32_t numOfChild,
SReadHandle* pHandle, SOperatorInfo** pOptrInfo) {
SExecTaskInfo* pTaskInfo, int32_t numOfChild, SReadHandle* pHandle,
SOperatorInfo** pOptrInfo) {
QRY_PARAM_CHECK(pOptrInfo);
int32_t code = TSDB_CODE_SUCCESS;
@ -4111,7 +4114,7 @@ int32_t createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhys
SOperatorInfo* pOperator = NULL;
code = createStreamSessionAggOperatorInfo(downstream, pPhyNode, pTaskInfo, pHandle, &pOperator);
if (pOperator == NULL || code != 0) {
downstream = NULL;
downstream = NULL;
QUERY_CHECK_CODE(code, lino, _error);
}
@ -4618,8 +4621,8 @@ int32_t doStreamStateDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
QUERY_CHECK_CODE(code, lino, _end);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
code =
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &winfo.sessionWin, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
code = tSimpleHashPut(pInfo->streamAggSup.pResultRows, &winfo.sessionWin, sizeof(SSessionKey), &winfo,
sizeof(SResultWindowInfo));
QUERY_CHECK_CODE(code, lino, _end);
}
@ -4846,7 +4849,7 @@ void streamStateReleaseState(SOperatorInfo* pOperator) {
int32_t resSize = winSize + sizeof(TSKEY);
char* pBuff = taosMemoryCalloc(1, resSize);
if (!pBuff) {
return ;
return;
}
memcpy(pBuff, pInfo->historyWins->pData, winSize);
memcpy(pBuff + winSize, &pInfo->twAggSup.maxTs, sizeof(TSKEY));
@ -4944,7 +4947,7 @@ void streamStateReloadState(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
}
}
taosMemoryFree(pBuf);
taosMemoryFreeClear(pBuf);
SOperatorInfo* downstream = pOperator->pDownstream[0];
if (downstream->fpSet.reloadStreamStateFn) {
@ -4953,6 +4956,7 @@ void streamStateReloadState(SOperatorInfo* pOperator) {
reloadAggSupFromDownStream(downstream, &pInfo->streamAggSup);
_end:
taosMemoryFreeClear(pBuf);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
}
@ -5001,9 +5005,9 @@ int32_t createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
pInfo->binfo.pRes = pResBlock;
SExprSupp* pExpSup = &pOperator->exprSupp;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = NULL;
SExprSupp* pExpSup = &pOperator->exprSupp;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = NULL;
code = createExprInfo(pStateNode->window.pFuncs, NULL, &pExprInfo, &numOfCols);
QUERY_CHECK_CODE(code, lino, _error);
@ -5335,7 +5339,7 @@ int32_t createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
*(pInfo->pState) = *(pTaskInfo->streamInfo.pState);
pAPI->stateStore.streamStateSetNumber(pInfo->pState, -1, pInfo->primaryTsIndex);
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
SExprInfo* pExprInfo = NULL;
code = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &pExprInfo, &numOfCols);
QUERY_CHECK_CODE(code, lino, _error);
@ -5383,7 +5387,8 @@ int32_t createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
pInfo->pState->pFileState = NULL;
code = pTaskInfo->storageAPI.stateStore.streamFileStateInit(
tsStreamBufferSize, sizeof(SWinKey), pInfo->aggSup.resultRowSize, funResSize, compareTs, pInfo->pState,
pInfo->twAggSup.deleteMark, GET_TASKID(pTaskInfo), pHandle->checkpointId, STREAM_STATE_BUFF_HASH, &pInfo->pState->pFileState);
pInfo->twAggSup.deleteMark, GET_TASKID(pTaskInfo), pHandle->checkpointId, STREAM_STATE_BUFF_HASH,
&pInfo->pState->pFileState);
QUERY_CHECK_CODE(code, lino, _error);
pInfo->pOperator = pOperator;
@ -5397,7 +5402,7 @@ int32_t createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode*
pInfo->recvGetAll = false;
code = createSpecialDataBlock(STREAM_CHECKPOINT, &pInfo->pCheckpointRes);
QUERY_CHECK_CODE(code, lino, _error);
QUERY_CHECK_CODE(code, lino, _error);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pInfo->pDeletedMap = tSimpleHashInit(4096, hashFn);

9
source/libs/parser/src/parTranslater.c
View File

@ -9581,6 +9581,10 @@ static int32_t translateDropUser(STranslateContext* pCxt, SDropUserStmt* pStmt)
static int32_t translateCreateAnode(STranslateContext* pCxt, SCreateAnodeStmt* pStmt) {
SMCreateAnodeReq createReq = {0};
createReq.urlLen = strlen(pStmt->url) + 1;
if (createReq.urlLen > TSDB_ANAL_ANODE_URL_LEN) {
return TSDB_CODE_MND_ANODE_TOO_LONG_URL;
}
createReq.url = taosMemoryCalloc(createReq.urlLen, 1);
if (createReq.url == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
@ -10817,6 +10821,11 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStm
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported Group by");
}
if (NULL == pSelect->pWindow && pSelect->hasAggFuncs) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
"Non window query only support scalar function, aggregate function is not allowed");
}
return TSDB_CODE_SUCCESS;
}

8
source/libs/stream/src/streamDispatch.c
View File

@ -706,9 +706,13 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
if (pTask->subtableWithoutMd5 != 1 && !isAutoTableName(pDataBlock->info.parTbName) &&
!alreadyAddGroupId(pDataBlock->info.parTbName, groupId) && groupId != 0) {
if (pTask->ver == SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
buildCtbNameAddGroupId(NULL, pDataBlock->info.parTbName, groupId);
code = buildCtbNameAddGroupId(NULL, pDataBlock->info.parTbName, groupId, sizeof(pDataBlock->info.parTbName));
} else if (pTask->ver > SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
buildCtbNameAddGroupId(pTask->outputInfo.shuffleDispatcher.stbFullName, pDataBlock->info.parTbName, groupId);
code = buildCtbNameAddGroupId(pTask->outputInfo.shuffleDispatcher.stbFullName, pDataBlock->info.parTbName,
groupId, sizeof(pDataBlock->info.parTbName));
}
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
} else {

79
source/util/src/tanal.c
View File

@ -76,7 +76,11 @@ int32_t taosAnalInit() {
}
tsAlgos.ver = 0;
taosThreadMutexInit(&tsAlgos.lock, NULL);
if (taosThreadMutexInit(&tsAlgos.lock, NULL) != 0) {
uError("failed to init algo mutex");
return -1;
}
tsAlgos.hash = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
if (tsAlgos.hash == NULL) {
uError("failed to init algo hash");
@ -99,7 +103,9 @@ static void taosAnalFreeHash(SHashObj *hash) {
void taosAnalCleanup() {
curl_global_cleanup();
taosThreadMutexDestroy(&tsAlgos.lock);
if (taosThreadMutexDestroy(&tsAlgos.lock) != 0) {
uError("failed to destroy anal lock");
}
taosAnalFreeHash(tsAlgos.hash);
tsAlgos.hash = NULL;
uInfo("analysis env is cleaned up");
@ -107,12 +113,15 @@ void taosAnalCleanup() {
void taosAnalUpdate(int64_t newVer, SHashObj *pHash) {
if (newVer > tsAlgos.ver) {
taosThreadMutexLock(&tsAlgos.lock);
SHashObj *hash = tsAlgos.hash;
tsAlgos.ver = newVer;
tsAlgos.hash = pHash;
taosThreadMutexUnlock(&tsAlgos.lock);
taosAnalFreeHash(hash);
if (taosThreadMutexLock(&tsAlgos.lock) == 0) {
SHashObj *hash = tsAlgos.hash;
tsAlgos.ver = newVer;
tsAlgos.hash = pHash;
if (taosThreadMutexUnlock(&tsAlgos.lock) != 0) {
uError("failed to unlock hash")
}
taosAnalFreeHash(hash);
}
} else {
taosAnalFreeHash(pHash);
}
@ -158,18 +167,22 @@ int32_t taosAnalGetAlgoUrl(const char *algoName, EAnalAlgoType type, char *url,
char name[TSDB_ANAL_ALGO_KEY_LEN] = {0};
int32_t nameLen = 1 + snprintf(name, sizeof(name) - 1, "%d:%s", type, algoName);
taosThreadMutexLock(&tsAlgos.lock);
SAnalUrl *pUrl = taosHashAcquire(tsAlgos.hash, name, nameLen);
if (pUrl != NULL) {
tstrncpy(url, pUrl->url, urlLen);
uDebug("algo:%s, type:%s, url:%s", algoName, taosAnalAlgoStr(type), url);
} else {
url[0] = 0;
terrno = TSDB_CODE_ANAL_ALGO_NOT_FOUND;
code = terrno;
uError("algo:%s, type:%s, url not found", algoName, taosAnalAlgoStr(type));
if (taosThreadMutexLock(&tsAlgos.lock) == 0) {
SAnalUrl *pUrl = taosHashAcquire(tsAlgos.hash, name, nameLen);
if (pUrl != NULL) {
tstrncpy(url, pUrl->url, urlLen);
uDebug("algo:%s, type:%s, url:%s", algoName, taosAnalAlgoStr(type), url);
} else {
url[0] = 0;
terrno = TSDB_CODE_ANAL_ALGO_NOT_FOUND;
code = terrno;
uError("algo:%s, type:%s, url not found", algoName, taosAnalAlgoStr(type));
}
if (taosThreadMutexUnlock(&tsAlgos.lock) != 0) {
uError("failed to unlock hash");
return TSDB_CODE_OUT_OF_MEMORY;
}
}
taosThreadMutexUnlock(&tsAlgos.lock);
return code;
}
@ -210,10 +223,10 @@ static int32_t taosCurlGetRequest(const char *url, SCurlResp *pRsp) {
return -1;
}
curl_easy_setopt(curl, CURLOPT_URL, url);
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, taosCurlWriteData);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, pRsp);
curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, 100);
if (curl_easy_setopt(curl, CURLOPT_URL, url) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, taosCurlWriteData) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_WRITEDATA, pRsp) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, 100) != 0) goto _OVER;
uDebug("curl get request will sent, url:%s", url);
code = curl_easy_perform(curl);
@ -238,14 +251,14 @@ static int32_t taosCurlPostRequest(const char *url, SCurlResp *pRsp, const char
}
headers = curl_slist_append(headers, "Content-Type:application/json;charset=UTF-8");
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
curl_easy_setopt(curl, CURLOPT_URL, url);
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, taosCurlWriteData);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, pRsp);
curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, 60000);
curl_easy_setopt(curl, CURLOPT_POST, 1);
curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, bufLen);
curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf);
if (curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_URL, url) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, taosCurlWriteData) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_WRITEDATA, pRsp) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, 60000) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_POST, 1) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, bufLen) != 0) goto _OVER;
if (curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf) != 0) goto _OVER;
uDebug("curl post request will sent, url:%s len:%d", url, bufLen);
code = curl_easy_perform(curl);
@ -596,7 +609,9 @@ void taosAnalBufDestroy(SAnalBuf *pBuf) {
SAnalColBuf *pCol = &pBuf->pCols[i];
if (pCol->fileName[0] != 0) {
if (pCol->filePtr != NULL) (void)taosCloseFile(&pCol->filePtr);
taosRemoveFile(pCol->fileName);
if (taosRemoveFile(pCol->fileName) != 0) {
uError("failed to remove file %s", pCol->fileName);
}
pCol->fileName[0] = 0;
}
}

1
source/util/src/terror.c
View File

@ -349,6 +349,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_ALREADY_EXIST, "Anode already exists"
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_NOT_EXIST, "Anode not there")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_TOO_LONG_URL, "Anode too long url")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_INVALID_PROTOCOL, "Anode invalid protocol")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_INVALID_ALGO_TYPE, "Anode invalid algorithm type")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_INVALID_VERSION, "Anode invalid version")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_TOO_MANY_ALGO, "Anode too many algorithm")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME, "Anode too long algorithm name")

13
source/util/src/tjson.c
View File

@ -207,6 +207,19 @@ int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal) {
return TSDB_CODE_SUCCESS;
}
int32_t tjsonGetStringValue2(const SJson* pJson, const char* pName, char* pVal, int32_t maxLen) {
char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName));
if (NULL == p) {
return TSDB_CODE_SUCCESS;
}
int32_t len = strlen(p);
if (len >= maxLen-1) {
return TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pVal, p);
return TSDB_CODE_SUCCESS;
}
int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal) {
char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName));
if (NULL == p) {

3
tests/parallel_test/cases.task
View File

@ -6,8 +6,6 @@
,,n,unit-test,bash test.sh
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py
#
# army-test
#
@ -1543,6 +1541,7 @@
,,n,develop-test,python3 ./test.py -f 2-query/ts-range.py
,,n,develop-test,python3 ./test.py -f 2-query/tag_scan.py
,,n,develop-test,python3 ./test.py -f 2-query/show_create_db.py
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/custom_col_tag.py
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/default_json.py
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/demo.py