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

Merge branch '3.0' into fix/3_liaohj

This commit is contained in:
Haojun Liao 2024-09-24 18:36:18 +08:00
parent 2b78b660bd ec143691ef
commit 4e4003c5a7
306 changed files with 9799 additions and 2456 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cmake/addr2line_CMakeLists.txt.in
View File

@ -5,8 +5,9 @@ ExternalProject_Add(addr2line
GIT_TAG master
SOURCE_DIR "${TD_CONTRIB_DIR}/addr2line"
BINARY_DIR "${TD_CONTRIB_DIR}/addr2line"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

3
cmake/apr-util_CMakeLists.txt.in
View File

@ -11,9 +11,10 @@ ExternalProject_Add(aprutil-1
BUILD_IN_SOURCE TRUE
BUILD_ALWAYS 1
#UPDATE_COMMAND ""
CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1/ --with-apr=$ENV{HOME}/.cos-local.1
CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1/ --with-apr=$ENV{HOME}/.cos-local.1
#CONFIGURE_COMMAND ./configure --with-apr=/usr/local/apr
BUILD_COMMAND make
INSTALL_COMMAND make install
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/apr_CMakeLists.txt.in
View File

@ -16,4 +16,5 @@ ExternalProject_Add(apr-1
BUILD_COMMAND make
INSTALL_COMMAND make install
TEST_COMMAND ""
GIT_SHALLOW true
)

3
cmake/cjson_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(cjson
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

1
cmake/cos_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(cos
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

5
cmake/crashdump_CMakeLists.txt.in
View File

@ -5,8 +5,9 @@ ExternalProject_Add(crashdump
GIT_TAG master
SOURCE_DIR "${TD_CONTRIB_DIR}/crashdump"
BINARY_DIR "${TD_CONTRIB_DIR}/crashdump"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/curl_CMakeLists.txt.in
View File

@ -16,4 +16,5 @@ ExternalProject_Add(curl2
BUILD_COMMAND make -j
INSTALL_COMMAND make install
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/geos_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(geos
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/gnuregex_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(gnuregex
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

3
cmake/gtest_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(googletest
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

3
cmake/iconv_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(iconv
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

5
cmake/libdwarf_CMakeLists.txt.in
View File

@ -5,8 +5,9 @@ ExternalProject_Add(libdwarf
GIT_TAG libdwarf-0.3.1
SOURCE_DIR "${TD_CONTRIB_DIR}/libdwarf"
BINARY_DIR "${TD_CONTRIB_DIR}/libdwarf"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/libs3_CMakeLists.txt.in
View File

@ -13,4 +13,5 @@ ExternalProject_Add(libs3
BUILD_COMMAND make build/lib/libs3.a
INSTALL_COMMAND make install_static
TEST_COMMAND ""
GIT_SHALLOW true
)

5
cmake/libuv_CMakeLists.txt.in
View File

@ -5,8 +5,9 @@ ExternalProject_Add(libuv
GIT_TAG v1.48.0
SOURCE_DIR "${TD_CONTRIB_DIR}/libuv"
BINARY_DIR "${TD_CONTRIB_DIR}/libuv"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

3
cmake/lz4_CMakeLists.txt.in
View File

@ -9,4 +9,5 @@ ExternalProject_Add(lz4
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

7
cmake/lzma_CMakeLists.txt.in
View File

@ -1,9 +1,9 @@
# xz
# xz
if (${TD_LINUX})
ExternalProject_Add(lzma2
GIT_REPOSITORY https://github.com/conor42/fast-lzma2.git
GIT_REPOSITORY https://github.com/conor42/fast-lzma2.git
SOURCE_DIR "${TD_CONTRIB_DIR}/lzma2"
#BINARY_DIR ""
BUILD_IN_SOURCE TRUE
@ -11,5 +11,6 @@ ExternalProject_Add(lzma2
BUILD_COMMAND make
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)
endif()
endif()

3
cmake/msvcregex_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(msvcregex
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

1
cmake/mxml_CMakeLists.txt.in
View File

@ -11,4 +11,5 @@ ExternalProject_Add(mxml
BUILD_COMMAND make
INSTALL_COMMAND make install
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/pcre2_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(pcre2
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/pthread_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(pthread
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

2
cmake/rocksdb_CMakeLists.txt.in
View File

@ -11,6 +11,7 @@ if (${BUILD_CONTRIB})
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)
else()
if (NOT ${TD_LINUX})
@ -24,6 +25,7 @@ else()
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)
endif()
endif()

3
cmake/sqlite_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(sqlite
BUILD_COMMAND "$(MAKE)"
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

1
cmake/ssl_CMakeLists.txt.in
View File

@ -12,4 +12,5 @@ ExternalProject_Add(openssl
BUILD_COMMAND make -j
INSTALL_COMMAND make install_sw -j
TEST_COMMAND ""
GIT_SHALLOW true
)

1
cmake/stub_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(stub
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

3
cmake/wcwidth_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(wcwidth
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

3
cmake/wingetopt_CMakeLists.txt.in
View File

@ -10,4 +10,5 @@ ExternalProject_Add(wingetopt
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

1
cmake/xml2_CMakeLists.txt.in
View File

@ -15,4 +15,5 @@ ExternalProject_Add(xml2
BUILD_COMMAND make -j
INSTALL_COMMAND make install && ln -sf $ENV{HOME}/.cos-local.2/include/libxml2/libxml $ENV{HOME}/.cos-local.2/include/libxml
TEST_COMMAND ""
GIT_SHALLOW true
)

9
cmake/xz_CMakeLists.txt.in
View File

@ -1,14 +1,15 @@
# xz
ExternalProject_Add(xz
GIT_REPOSITORY https://github.com/xz-mirror/xz.git
GIT_TAG v5.4.4
GIT_REPOSITORY https://github.com/xz-mirror/xz.git
GIT_TAG v5.4.4
SOURCE_DIR "${TD_CONTRIB_DIR}/xz"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
CMAKE_ARGS
CMAKE_ARGS
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
GIT_SHALLOW true
)

1
cmake/zlib_CMakeLists.txt.in
View File

@ -12,4 +12,5 @@ ExternalProject_Add(zlib
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
GIT_SHALLOW true
)

11
cmake/zstd_CMakeLists.txt.in
View File

@ -1,15 +1,16 @@
# zstb
# zstb
#ExternalProject_Add(zstd
#GIT_REPOSITORY https://github.com/facebook/zstd.git
#GIT_REPOSITORY https://github.com/facebook/zstd.git
#GIT_TAG v1.5.5
#SOURCE_DIR "${TD_CONTRIB_DIR}/zstd"
#DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
#BINARY_DIR ""
#CMAKE_COMMAND
#CMAKE_COMMAND
#CMAKE_ARGS ${TD_CONTRIB_DIR}/zstd/build/cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_ARCHIVE_OUTPUT_DIRECTORY=${CMAKE_BINARY_DIR}/build/lib
#BUILD_COMMAND make -j4
#BUILD_COMMAND make -j4
#INSTALL_COMMAND ""
#TEST_COMMAND ""
#)
#GIT_SHALLOW true
#)

4
docs/zh/01-index.md
View File

@ -8,7 +8,7 @@ TDengine 是一款[开源](https://www.taosdata.com/tdengine/open_source_time-se
TDengine 充分利用了时序数据的特点，提出了“一个数据采集点一张表”与“超级表”的概念，设计了创新的存储引擎，让数据的写入、查询和存储效率都得到极大的提升。为正确理解并使用 TDengine，无论如何，请您仔细阅读[快速入门](./basic)一章。
如果你是开发工程师，请一定仔细阅读[开发指南](./develop)一章，该部分对数据库连接、建模、插入数据、查询、流式计算、缓存、数据订阅、用户自定义函数等功能都做了详细介绍，并配有各种编程语言的示例代码。大部分情况下，你只要复制粘贴示例代码，针对自己的应用稍作改动，就能跑起来。
如果你是开发工程师，请一定仔细阅读[开发指南](./develop)一章，该部分对数据库连接、建模、插入数据、查询、流式计算、缓存、数据订阅、用户自定义函数等功能都做了详细介绍，并配有各种编程语言的示例代码。大部分情况下，你只要复制粘贴示例代码，针对自己的应用稍作改动，就能跑起来。对 REST API、各种编程语言的连接器（Connector）想做更多详细了解的话，请看[连接器](./reference/connector)一章。
我们已经生活在大数据时代，纵向扩展已经无法满足日益增长的业务需求，任何系统都必须具有水平扩展的能力，集群成为大数据以及 Database 系统的不可缺失功能。TDengine 团队不仅实现了集群功能，而且将这一重要核心功能开源。怎么部署、管理和维护 TDengine 集群，请仔细参考[运维管理](./operation)一章。
@ -16,8 +16,6 @@ TDengine 采用 SQL 作为查询语言，大大降低学习成本、降低迁移
如果你是系统管理员，关心安装、升级、容错灾备、关心数据导入、导出、配置参数，如何监测 TDengine 是否健康运行，如何提升系统运行的性能，请仔细参考[运维指南](./operation)一章。
如果你对 TDengine 的产品组件、REST API、各种编程语言的连接器（Connector）想做更多详细了解，请看[参考指南](./reference)一章。
最后，作为一个开源软件，欢迎大家的参与。如果发现文档有任何错误、描述不清晰的地方，请在每个页面的最下方，点击“编辑本文档”直接进行修改。
Together, we make a difference!

2
docs/zh/06-advanced/03-stream.md
View File

@ -72,7 +72,7 @@ SELECT _wstart, count(*), avg(voltage) FROM power.meters PARTITION BY tbname INT
- stb_name 是保存计算结果的超级表的表名，如果该超级表不存在，则会自动创建；如果已存在，则检查列的 schema 信息。详见 6.3.8 节。
- tags 子句定义了流计算中创建标签的规则。通过 tags 字段可以为每个分区对应的子表生成自定义的标签值。
## 流试计算的规则和策略
## 流式计算的规则和策略
### 流计算的分区

2
docs/zh/06-advanced/05-data-in/07-mqtt.md
View File

@ -55,7 +55,7 @@ TDengine 可以通过 MQTT 连接器从 MQTT 代理订阅数据并将其写入 T
在 **MQTT 协议** 下拉列表中选择 MQTT 协议版本。有三个选项：`3.1`、`3.1.1`、`5.0`。 默认值为 3.1。
在 **Client ID** 中填写客户端 ID。建多个任务时，同个 MQTT 地址下，Client ID 必须是唯一的。
在 **Client ID** 中填写客户端标识，填写后会生成带有 `taosx` 前缀的客户端 id （例如，如果填写的标识为 `foo`，则生成的客户端 id 为 `taosxfoo`）。如果打开末尾处的开关，则会把当前任务的任务 id 拼接到 `taosx` 之后，输入的标识之前（生成的客户端 id 形如 `taosx100foo`）。连接到同一个 MQTT 地址的所有客户端 id 必须保证唯一。
在 **Keep Alive** 中输入保持活动间隔。如果代理在保持活动间隔内没有收到来自客户端的任何消息，它将假定客户端已断开连接，并关闭连接。
保持活动间隔是指客户端和代理之间协商的时间间隔，用于检测客户端是否活动。如果客户端在保持活动间隔内没有向代理发送消息，则代理将断开连接。

4
docs/zh/06-advanced/05-data-in/08-kafka.md
View File

@ -60,6 +60,10 @@ TDengine 可以高效地从 Kafka 读取数据并将其写入 TDengine，以实
在 **主题** 中填写要消费的 Topic 名称。可以配置多个 Topic ， Topic 之间用逗号分隔。例如：`tp1,tp2`。
在 **Client ID** 中填写客户端标识，填写后会生成带有 `taosx` 前缀的客户端 ID （例如，如果填写的标识为 `foo`，则生成的客户端 ID 为 `taosxfoo`）。如果打开末尾处的开关，则会把当前任务的任务 ID 拼接到 `taosx` 之后，输入的标识之前（生成的客户端 ID 形如 `taosx100foo`）。连接到同一个 Kafka 集群的所有客户端 ID 必须保证唯一。
在 **消费者组 ID** 中填写消费者组标识，填写后会生成带有 `taosx` 前缀的消费者组 ID （例如，如果填写的标识为 `foo`，则生成的消费者组 ID 为 `taosxfoo`）。如果打开末尾处的开关，则会把当前任务的任务 ID 拼接到 `taosx` 之后，输入的标识之前（生成的消费者组 ID 形如 `taosx100foo`）。
在 **Offset** 的下拉列表中选择从哪个 Offset 开始消费数据。有三个选项：`Earliest`、`Latest`、`ByTime(ms)`。 默认值为Earliest。
* Earliest：用于请求最早的 offset。

BIN
docs/zh/06-advanced/05-data-in/kafka-06.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 17 KiB

After

Width:  |  Height:  |  Size: 48 KiB

BIN
docs/zh/06-advanced/05-data-in/mqtt-05.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 105 KiB

After

Width:  |  Height:  |  Size: 37 KiB

10
docs/zh/07-operation/06-gui.md
View File

@ -229,4 +229,12 @@ toc_max_heading_level: 4
- 权限
- 白名单
![management-01-importInfo.jpeg](./pic/management-01-importInfo.jpeg)。
![management-01-importInfo.jpeg](./pic/management-01-importInfo.jpeg)
### 慢 SQL
点击“系统管理”后，点击“慢 SQL”标签页，可以查看慢 SQL 执行语句日志统计与明细。
- 慢 SQL 明细：默认展示的是开始执行时间是一天内和执行耗时大于等于10秒的数据
![management-01-slowsql.jpeg](./pic/management-01-slowsql.jpeg)
- 慢 SQL 统计：默认展示所有的数据，可根据开始执行时间进行过滤
![management-02-slowsql.jpeg](./pic/management-02-slowsql.jpeg)

26
docs/zh/07-operation/12-multi.md
View File

@ -69,7 +69,7 @@ dataDir /mnt/data6 2 0
在配置文件 /etc/taos/taos.cfg 中，添加用于 S3 访问的参数：
|参数名称 | 参数含义 |
|:-------------:|:-----------------------------------------------:|
|:-------------|:-----------------------------------------------|
|s3EndPoint | 用户所在地域的 COS 服务域名，支持 http 和 https，bucket 的区域需要与 endpoint 的保持一致，否则无法访问。例如：http://cos.ap-beijing.myqcloud.com |
|s3AccessKey |冒号分隔的用户 SecretId:SecretKey。例如：AKIDsQmwsfKxTo2A6nGVXZN0UlofKn6JRRSJ:lIdoy99ygEacU7iHfogaN2Xq0yumSm1E |
|s3BucketName | 存储桶名称，减号后面是用户注册 COS 服务的 AppId。其中 AppId 是 COS 特有，AWS 和阿里云都没有，配置时需要作为 bucket name 的一部分，使用减号分隔。参数值均为字符串类型，但不需要引号。例如：test0711-1309024725 |
@ -111,3 +111,27 @@ s3migrate database <db_name>;
| 1 | s3_keeplocal | 3650 | 1 | 365000 | 数据在本地保留的天数，即 data 文件在本地磁盘保留多长时间后可以上传到 S3。默认单位：天，支持 m（分钟）、h（小时）和 d（天）三个单位 |
| 2 | s3_chunksize | 262144 | 131072 | 1048576 | 上传对象的大小阈值，与 TSDB_PAGESIZE 参数一样，不可修改，单位为 TSDB 页 |
| 3 | s3_compact | 0 | 0 | 1 | TSDB 文件组首次上传 S3 时，是否自动进行 compact 操作。 |
## Azure Blob 存储
本节介绍在 TDengine Enterprise 如何使用微软 Azure Blob 对象存储。本功能是上一小节‘对象存储’功能的扩展，需额外依赖 Flexify 服务提供的 S3 网关。通过适当的参数配置，可以把大部分较冷的时序数据存储到 Azure Blob 服务中。
### Flexify 服务
Flexify 是 Azure Marketplace 中的一款应用程序，允许兼容 S3 的应用程序通过标准 S3 API 在 Azure Blob Storage 中存储数据。可使用多个 Flexify 服务对同一个 Blob 存储建立多个 S3 网关。
部署方式请参考 [Flexify](https://azuremarketplace.microsoft.com/en-us/marketplace/apps/flexify.azure-s3-api?tab=Overview) 应用页面说明。
### 配置方式
在配置文件 /etc/taos/taos.cfg 中，添加用于 S3 访问的参数：
```
s3EndPoint http //20.191.157.23,http://20.191.157.24,http://20.191.157.25
s3AccessKey FLIOMMNL0:uhRNdeZMLD4wo,ABCIOMMN:uhRNdeZMD4wog,DEFOMMNL049ba:uhRNdeZMLD4wogXd
s3BucketName td-test
```
- 允许对 s3EndPoint、s3AccessKey 配置多项，但要求二者项数一致。多个配置项间使用 ',' 分隔。s3BucketName 仅允许配置一项
- 认为每一组 `{s3EndPoint、s3AccessKey}` 配置对应一个 S3 服务，每次发起 S3 请求时将随机选择一个服务
- 认为全部 S3 服务均指向同一数据源，对各个 S3 服务操作完全等价
- 在某一 S3 服务上操作失败后会切换至其他服务，全部服务都失败后将返回最后产生的错误码
- 最大支持的 S3 服务配置数为 10

BIN
docs/zh/07-operation/pic/management-01-slowsql.jpeg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 283 KiB

BIN
docs/zh/07-operation/pic/management-02-slowsql.jpeg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 256 KiB

18
docs/zh/08-develop/01-connect/index.md
View File

@ -132,7 +132,7 @@ TDengine 提供了丰富的应用程序开发接口，为了便于用户快速
pip3 install taospy[ws]
```
- **安装验证**
- **安装验证**
<Tabs defaultValue="rest">
<TabItem value="native" label="原生连接">
对于原生连接，需要验证客户端驱动和 Python 连接器本身是否都正确安装。如果能成功导入 `taos` 模块，则说明已经正确安装了客户端驱动和 Python 连接器。可在 Python 交互式 Shell 中输入：
@ -198,18 +198,18 @@ taos = { version = "*", default-features = false, features = ["ws"] }
- **安装**
- 使用 npm 安装 Node.js 连接器
```
npm install @tdengine/websocket
```
```
npm install @tdengine/websocket
```
:::note Node.js 目前只支持 Websocket 连接
- **安装验证**
- 新建安装验证目录，例如：`~/tdengine-test`，下载 GitHub 上 [nodejsChecker.js 源代码](https://github.com/taosdata/TDengine/tree/main/docs/examples/node/websocketexample/nodejsChecker.js)到本地。
- 在命令行中执行以下命令。
```bash
npm init -y
npm install @tdengine/websocket
node nodejsChecker.js
```
```bash
npm init -y
npm install @tdengine/websocket
node nodejsChecker.js
```
- 执行以上步骤后，在命令行会输出 nodeChecker.js 连接 TDengine 实例，并执行简单插入和查询的结果。

6
docs/zh/08-develop/07-tmq.md
View File

@ -34,13 +34,15 @@ TDengine 消费者的概念跟 Kafka 类似，消费者通过订阅主题来接
| `td.connect.user` | string | 用户名 | |
| `td.connect.pass` | string | 密码 | |
| `td.connect.port` | integer | 服务端的端口号 | |
| `group.id` | string | 消费组 ID，同一消费组共享消费进度 | <br />**必填项**。最大长度：192。<br />每个topic最多可建立100个 consumer group |
| `client.id` | string | 客户端 ID | 最大长度：192。 |
| `group.id` | string | 消费组 ID，同一消费组共享消费进度 | <br />**必填项**。最大长度：192。<br />每个topic最多可建立 100 个 consumer group |
| `client.id` | string | 客户端 ID | 最大长度：192 |
| `auto.offset.reset` | enum | 消费组订阅的初始位置 | <br />`earliest`: default(version < 3.2.0.0);从头开始订阅; <br/>`latest`: default(version >= 3.2.0.0);仅从最新数据开始订阅; <br/>`none`: 没有提交的 offset 无法订阅 |
| `enable.auto.commit` | boolean | 是否启用消费位点自动提交，true: 自动提交，客户端应用无需commit；false：客户端应用需要自行commit | 默认值为 true |
| `auto.commit.interval.ms` | integer | 消费记录自动提交消费位点时间间隔，单位为毫秒 | 默认值为 5000 |
| `msg.with.table.name` | boolean | 是否允许从消息中解析表名, 不适用于列订阅（列订阅时可将 tbname 作为列写入 subquery 语句）（从3.2.0.0版本该参数废弃，恒为true） | 默认关闭 |
| `enable.replay` | boolean | 是否开启数据回放功能 | 默认关闭 |
| `session.timeout.ms` | integer | consumer 心跳丢失后超时时间，超时后会触发 rebalance 逻辑，成功后该 consumer 会被删除（从3.3.3.0版本开始支持） | 默认值为 12000，取值范围 [6000， 1800000] |
| `max.poll.interval.ms` | integer | consumer poll 拉取数据间隔的最长时间，超过该时间，会认为该 consumer 离线，触发rebalance 逻辑，成功后该 consumer 会被删除（从3.3.3.0版本开始支持） | 默认值为 300000，[1000，INT32_MAX] |
下面是各语言连接器创建参数：

3
docs/zh/14-reference/01-components/02-taosc.md Normal file → Executable file
View File

@ -38,7 +38,8 @@ TDengine 客户端驱动提供了应用编程所需要的全部 API，并且在
|enableCoreFile | crash 时是否生成 core 文件，0: 不生成， 1： 生成；缺省值：1 |
|enableScience | 是否开启科学计数法显示浮点数; 0: 不开始, 1: 开启；缺省值：1 |
|compressMsgSize | 是否对 RPC 消息进行压缩; -1: 所有消息都不压缩; 0: 所有消息都压缩; N (N>0): 只有大于 N 个字节的消息才压缩; 缺省值 -1|
|queryTableNotExistAsEmpty | 查询表不存在时是否返回空结果集; false: 返回错误; true: 返回空结果集; 缺省值 false|
## API
请参考[连接器](../../connector)
请参考[连接器](../../connector)

4
docs/zh/14-reference/01-components/03-taosadapter.md
View File

@ -187,8 +187,8 @@ AllowWebSockets
- 兼容 InfluxDB v1 写接口
[https://docs.influxdata.com/influxdb/v2.0/reference/api/influxdb-1x/write/](https://docs.influxdata.com/influxdb/v2.0/reference/api/influxdb-1x/write/)
- 兼容 OpenTSDB JSON 和 telnet 格式写入
- \<http://opentsdb.net/docs/build/html/api_http/put.html\>
- \<http://opentsdb.net/docs/build/html/api_telnet/put.html\>
- [http://opentsdb.net/docs/build/html/api_http/put.html](http://opentsdb.net/docs/build/html/api_http/put.html)
- [http://opentsdb.net/docs/build/html/api_telnet/put.html](http://opentsdb.net/docs/build/html/api_telnet/put.html)
- 与 collectd 无缝连接。
collectd 是一个系统统计收集守护程序，请访问 [https://collectd.org/](https://collectd.org/) 了解更多信息。
- Seamless connection with StatsD。

2
docs/zh/14-reference/01-components/06-taoskeeper.md
View File

@ -194,7 +194,7 @@ Active: inactive (dead)
可以访问 taosKeeper 的 `check_health` 接口来判断服务是否存活，如果服务正常则会返回 HTTP 200 状态码：
```
$ curl -i http://127.0.0.1:6043/check_health
curl -i http://127.0.0.1:6043/check_health
```
返回结果：

2
docs/zh/14-reference/03-taos-sql/12-distinguished.md
View File

@ -54,7 +54,7 @@ window_clause: {
```
其中，interval_val 和 sliding_val 都表示时间段，interval_offset 表示窗口偏移量，interval_offset 必须小于 interval_val，语法上支持三种方式，举例说明如下:
- INTERVAL(1s, 500a) SLIDING(1s), 自带时间单位的形式，其中的时间单位是单字符表示, 分别为: a (毫秒), b (纳秒), d (天), h (小时), m (分钟), n (月), s (秒), u (微妙), w (周), y (年).
- INTERVAL(1s, 500a) SLIDING(1s), 自带时间单位的形式，其中的时间单位是单字符表示, 分别为: a (毫秒), b (纳秒), d (天), h (小时), m (分钟), n (月), s (秒), u (微秒), w (周), y (年).
- INTERVAL(1000, 500) SLIDING(1000), 不带时间单位的形式，将使用查询库的时间精度作为默认时间单位，当存在多个库时默认采用精度更高的库.
- INTERVAL('1s', '500a') SLIDING('1s'), 自带时间单位的字符串形式，字符串内部不能有任何空格等其它字符.

6
docs/zh/14-reference/03-taos-sql/22-meta.md
View File

@ -296,8 +296,10 @@ TDengine 内置了一个名为 `INFORMATION_SCHEMA` 的数据库，提供对数
| 2 | consumer_group | VARCHAR(193) | 订阅者的消费者组 |
| 3 | vgroup_id | INT | 消费者被分配的 vgroup id |
| 4 | consumer_id | BIGINT | 消费者的唯一 id |
| 5 | offset | VARCHAR(64) | 消费者的消费进度 |
| 6 | rows | BIGINT | 消费者的消费的数据条数 |
| 5 | user | VARCHAR(24) | 消费者的登录的用户名 |
| 6 | fqdn | VARCHAR(128) | 消费者的所在机器的 fqdn |
| 7 | offset | VARCHAR(64) | 消费者的消费进度 |
| 8 | rows | BIGINT | 消费者的消费的数据条数 |
## INS_STREAMS

4
docs/zh/14-reference/03-taos-sql/28-index.md
View File

@ -18,7 +18,7 @@ time_duration:
number unit
```
创建 TSMA 时需要指定 TSMA 名字, 表名字, 函数列表以及窗口大小. 当基于一个已经存在的 TSMA 创建新的 TSMA 时, 需要使用 `RECURSIVE` 关键字但不能指定 `FUNCTION()`， 新创建的 TSMA 已有 TSMA 拥有相同的函数列表， 且此种情况下所指定的 INTERVAL 必须为所基于的 TSMA 窗口长度的整数倍。
创建 TSMA 时需要指定 TSMA 名字, 表名字, 函数列表以及窗口大小. 当基于一个已经存在的 TSMA 创建新的 TSMA 时, 需要使用 `RECURSIVE` 关键字但不能指定 `FUNCTION()`， 新创建的 TSMA 已有 TSMA 拥有相同的函数列表， 且此种情况下所指定的 INTERVAL 必须至少为所基于的 TSMA 窗口长度的整数倍, 并且天不能基于2h或3h建立, 只能基于1h建立, 月也只能基于1d而非2d,3d建立。
其中 TSMA 命名规则与表名字类似, 长度最大限制为表名长度限制减去输出表后缀长度, 表名长度限制为193, 输出表后缀为`_tsma_res_stb_`, TSMA 名字最大长度为178.
@ -28,7 +28,7 @@ TSMA只能基于超级表和普通表创建, 不能基于子表创建.
由于TSMA输出为一张超级表, 因此输出表的行长度受最大行长度限制, 不同函数的`中间结果`大小各异, 一般都大于原始数据大小, 若输出表的行长度大于最大行长度限制, 将会报`Row length exceeds max length`错误. 此时需要减少函数个数或者将常用的函数进行分组拆分到多个TSMA中.
窗口大小的限制为[1m ~ 1h]. INTERVAL 的单位与查询中INTERVAL子句相同, 如 a (毫秒), b (纳秒), h (小时), m (分钟), s (秒), u (微妙).
窗口大小的限制为[1m ~ 1y/12n]. INTERVAL 的单位与查询中INTERVAL子句相同, 如 a (毫秒), b (纳秒), h (小时), m (分钟), s (秒), u (微妙), d (天), w(周), n(月), y(年).
TSMA为库内对象, 但名字全局唯一. 集群内一共可创建TSMA个数受参数`maxTsmaNum`限制, 参数默认值为3, 范围: [0-3]. 注意, 由于TSMA后台计算使用流计算, 因此每创建一条TSMA, 将会创建一条流, 因此能够创建的TSMA条数也受当前已经存在的流条数和最大可创建流条数限制.

2
docs/zh/14-reference/05-connector/45-php.mdx
View File

@ -8,7 +8,7 @@ toc_max_heading_level: 4
PHP 连接器依赖 TDengine 客户端驱动。
项目地址：\<https://github.com/Yurunsoft/php-tdengine>
项目地址：[https://github.com/Yurunsoft/php-tdengine](https://github.com/Yurunsoft/php-tdengine)
TDengine 服务端或客户端安装后，`taos.h` 位于：

58
docs/zh/14-reference/05-connector/60-rest-api.mdx
View File

@ -118,41 +118,55 @@ curl -L -u username:password -d "<SQL>" <ip>:<PORT>/rest/sql/[db_name][?tz=timez
### HTTP 响应码
从 `TDengine 3.0.3.0` 开始 `taosAdapter` 提供配置参数 `httpCodeServerError` 用来设置当 C 接口返回错误时是否返回非 200 的http状态码
默认情况下，`taosAdapter` 对大多数 C 接口调用出错时也会返回 200 响应码，但是 HTTP body 中包含错误信息。从 `TDengine 3.0.3.0` 开始 `taosAdapter` 提供配置参数 `httpCodeServerError` 用来设置当 C 接口返回错误时是否返回非 200 的 HTTP 响应码。无论是否设置此参数，响应 body 里都有详细的错误码和错误信息，具体请参考 [错误](../rest-api/#错误) 。
| **说明** | **httpCodeServerError false** | **httpCodeServerError true** |
|--------------------|-------------------------------|---------------------------------------|
| taos_errno() 返回 0 | 200 | 200 |
| taos_errno() 返回 非0 | 200（除鉴权错误） | 500 （除鉴权错误和 400/502 错误） |
| 参数错误 | 400 (仅处理 HTTP 请求 URL 参数错误) | 400 （处理 HTTP 请求 URL 参数错误和 taosd 返回错误） |
| 鉴权错误 | 401 | 401 |
| 接口不存在 | 404 | 404 |
| 集群不可用错误 | 502 | 502 |
| 系统资源不足 | 503 | 503 |
**当 httpCodeServerError 为 false 时：**
返回 400 的 C 错误码为：
| **分类说明** |**HTTP 响应码** |
|--------------------|-------------------------------|
| C 接口调用成功 | 200 | 200 |
| C 接口调用出错，且不是鉴权错误 | 200 |
| HTTP 请求 URL 参数错误 | 400 |
| C 接口调用鉴权错误 | 401 |
| 接口不存在 | 404 |
| 系统资源不足 | 503 |
- TSDB_CODE_TSC_SQL_SYNTAX_ERROR ( 0x0216)
- TSDB_CODE_TSC_LINE_SYNTAX_ERROR (0x021B)
**当 httpCodeServerError 为 true 时：**
| **分类说明** | **HTTP 响应码** |
|--------------------|-------------------------------|
| C 接口调用成功 | 200 |
| HTTP 请求 URL 参数错误和 C 接口调用参数解析错误 | 400 |
| C 接口调用鉴权错误 | 401 |
| 接口不存在 | 404 |
| C 接口调用网络不可用错误 | 502 |
| 系统资源不足 |503 |
| 其他 C 接口调用错误 | 500 |
C 接口参数解析相关错误码：
- TSDB_CODE_TSC_SQL_SYNTAX_ERROR (0x0216)
- TSDB_CODE_TSC_LINE_SYNTAX_ERROR (0x021B)
- TSDB_CODE_PAR_SYNTAX_ERROR (0x2600)
- TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE (0x060B)
- TSDB_CODE_TSC_VALUE_OUT_OF_RANGE (0x0224)
- TSDB_CODE_TSC_VALUE_OUT_OF_RANGE (0x0224)
- TSDB_CODE_PAR_INVALID_FILL_TIME_RANGE (0x263B)
返回 401 的错误码为：
C 接口鉴权相关错误码：
- TSDB_CODE_MND_USER_ALREADY_EXIST (0x0350)
- TSDB_CODE_MND_USER_NOT_EXIST ( 0x0351)
- TSDB_CODE_MND_INVALID_USER_FORMAT (0x0352)
- TSDB_CODE_MND_INVALID_PASS_FORMAT (0x0353)
- TSDB_CODE_MND_USER_ALREADY_EXIST (0x0350)
- TSDB_CODE_MND_USER_NOT_EXIST (0x0351)
- TSDB_CODE_MND_INVALID_USER_FORMAT (0x0352)
- TSDB_CODE_MND_INVALID_PASS_FORMAT (0x0353)
- TSDB_CODE_MND_NO_USER_FROM_CONN (0x0354)
- TSDB_CODE_MND_TOO_MANY_USERS (0x0355)
- TSDB_CODE_MND_INVALID_ALTER_OPER (0x0356)
- TSDB_CODE_MND_AUTH_FAILURE (0x0357)
返回 403 的错误码为：
C 接口网络不可用相关错误码：
- TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED (0x0020)
- TSDB_CODE_RPC_NETWORK_UNAVAIL (0x000B)
错误码和错误描述请参考[错误码](../../../reference/error-code)
### HTTP body 结构
@ -333,6 +347,8 @@ curl --location 'http://<fqdn>:<port>/rest/sql' \
- code：（`int`）错误码。
- desc：（`string`）错误描述。
错误码和错误描述请参考[错误码](../../../reference/error-code)
#### 返回key-value形式数据
当指定 url 参数 `row_with_meta=true` 时，返回的 data 中的数据会从数组的形式变成对象的形式，对象的 key 为列名，value 为数据，如下所示：

2
docs/zh/20-third-party/01-collection/09-emq-broker.md vendored
View File

@ -16,7 +16,7 @@ MQTT 是流行的物联网数据传输协议，[EMQX](https://github.com/emqx/em
## 安装并启动 EMQX
用户可以根据当前的操作系统，到 EMQX 官网下载安装包，并执行安装。下载地址如下：\<https://www.emqx.io/zh/downloads>。安装后使用 `sudo emqx start` 或 `sudo systemctl start emqx` 启动 EMQX 服务。
用户可以根据当前的操作系统，到 [EMQX 官网下载安装包](https://www.emqx.io/zh/downloads)，并执行安装。安装后使用 `sudo emqx start` 或 `sudo systemctl start emqx` 启动 EMQX 服务。
注意：本文基于 EMQX v4.4.5 版本，其他版本由于相关配置界面、配置方法以及功能可能随着版本升级有所区别。

22
docs/zh/26-tdinternal/07-topic.md
View File

@ -61,37 +61,37 @@ TDengine 会为 WAL 文件自动创建索引以支持快速随机访问。通过
![数据订阅架构图](./topicarch.png)
在客户端成功建立与服务器的连接之后，用户须首先指定消费组和主题，以创建相应的消费者实例。随后，客户端便会向服务器提交订阅请求。此刻，消费者的状态被标记为 modify，表示正处于配置阶段。消费者随后会定期向服务器发送请求，以检索并获取待消费的 vnode 列表，直至服务器为其分配相应的 vnode。一旦分配完成，消费者的状态便更新为 ready，标志着订阅流程的成功完成。此刻，客户端便可正式启动向 vnode 发送消费数据请求的过程。
在客户端成功建立与服务器的连接之后，用户须首先指定消费组和主题，以创建相应的消费者实例。随后，客户端便会向服务器提交订阅请求。此刻，消费者的状态被标记为 rebalancing，表示正处于 rebalance 阶段。消费者随后会定期向服务器发送请求，以检索并获取待消费的 vnode 列表，直至服务器为其分配相应的 vnode。一旦分配完成，消费者的状态便更新为 ready，标志着订阅流程的成功完成。此刻，客户端便可正式启动向 vnode 发送消费数据请求的过程。
在消费数据的过程中，消费者会不断地向每个分配到的 vnode 发送请求，以尝试获取新的数据。一旦收到数据，消费者在完成消费后会继续向该 vnode 发送请求，以便持续消费。若在预设时间内未收到任何数据，消费者便会在 vnode 上注册一个消费 handle。这样一来，一旦 vnode 上有新数据产生，便会立即推送给消费者，从而确保数据消费的即时性，并有效减少消费者频繁主动拉取数据所导致的性能损耗。因此，可以看出，消费者从 vnode 获取数据的方式是一种拉取（pull）与推送（push）相结合的高效模式。
消费者在收到数据时，会同时收到数据的版本号，并将其记录为当前在每个 vnode上的消费进度。这一进度仅在消费者内部以内存形式存储，确保仅对该消费者有效。这种设计保证了消费者在每次启动时能够从上次的消费中断处继续，避免了数据的重复处理。然而，若消费者需要退出并希望在之后恢复上次的消费进度，就必须在退出前将消费进度提交至服务器，执行所谓的 commit 操作。这一操作会将消费进度在服务器进行持久化存储，支持自动提交或手动提交两种方式。
消费者在收到数据时，会同时收到数据的版本号，并将其记录为当前在每个 vnode 上的消费进度。这一进度仅在消费者内部以内存形式存储，确保仅对该消费者有效。这种设计保证了消费者在每次启动时能够从上次的消费中断处继续，避免了数据的重复处理。然而，若消费者需要退出并希望在之后恢复上次的消费进度，就必须在退出前将消费进度提交至服务器，执行所谓的 commit 操作。这一操作会将消费进度在服务器进行持久化存储，支持自动提交或手动提交两种方式。
此外，为了维持消费者的活跃状态，客户端还实施了心跳保活机制。通过定期向服务器发送心跳信号，消费者能够向服务器证明自己仍然在线。若服务器在一定时间内未收到消费者的心跳，便会将其标记为 lost 状态，即认为消费者已离线。服务器依赖心跳机制来监控所有消费者的状态，进而有效地管理整个消费者群体。
此外，为了维持消费者的活跃状态，客户端还实施了心跳保活机制。通过定期向服务器发送心跳信号，消费者能够向服务器证明自己仍然在线。若服务器在一定时间内未收到消费者的心跳，便会认为消费者已离线。对于一定时间（可以通过参数控制）不拉取数据的 consumer，服务器也会标记为离线，并从消费组中删除该消费者。服务器依赖心跳机制来监控所有消费者的状态，进而有效地管理整个消费者群体。
mnode 主要负责处理订阅过程中的控制消息，包括创建和删除主题、订阅消息、查询 endpoint 消息以及心跳消息等。vnode 则专注于处理消费消息和 commit 消息。当mnode 收到消费者的订阅消息时，如果该消费者尚未订阅过，其状态将被设置为 modify。如果消费者已经订阅过，但订阅的主题发生变更，消费者的状态同样会被设置为 modify。等待再平衡的计时器到来时，mnode 会对 modify 状态的消费者执行再平衡操作，将心跳超过固定时间的消费者设置为 lost 状态，并对关闭或 lost 状态超过一定时间的消费者进行删除操作。
mnode 主要负责处理订阅过程中的控制消息，包括创建和删除主题、订阅消息、查询 endpoint 消息以及心跳消息等。vnode 则专注于处理消费消息和 commit 消息。当 mnode 收到消费者的订阅消息时，如果该消费者尚未订阅过，其状态将被设置为 rebalancing。如果消费者已经订阅过，但订阅的主题发生变更，消费者的状态同样会被设置为 rebalancing。然后 mnode 会对 rebalancing 状态的消费者执行 rebalance 操作。心跳超过固定时间的消费者或主动关闭的消费者将被删除。
消费者会定期向 mnode 发送查询 endpoint 消息，以获取再平衡后的最新 vnode 分配结果。同时，消费者还会定期发送心跳消息，通知 mnode 自身处于活跃状态。此外，消费者的一些信息也会通过心跳消息上报至 mnode，用户可以查询 mnode 上的这些信息以监测各个消费者的状态。这种设计有助于实现对消费者的有效管理和监控。
## 再平衡过程
## rebalance 过程
每个主题的数据可能分散在多个 vnode 上。服务器通过执行再平衡过程，将这些vnode 合理地分配给各个消费者，确保数据的均匀分布和高效消费。
每个主题的数据可能分散在多个 vnode 上。服务器通过执行 rebalance 过程，将这些 vnode 合理地分配给各个消费者，确保数据的均匀分布和高效消费。
如下图所示，c1 表示消费者 1，c2 表示消费者 2，g1 表示消费组 1。起初 g1 中只有 c1 消费数据，c1 发送订阅信息到 mnode，mnode 把数据所在的所有 4 个vnode 分配给 c1。当 c2 增加到 g1 后，c2 将订阅信息发送给 mnode，mnode 将在下一次再平衡计时器到来时检测到这个 g1 需要重新分配，就会启动再平衡过程，随后 c2 分得其中两个 vnode 消费。分配信息还会通过 mnode 发送给 vnode，同时 c1 和 c2 也会获取自己消费的 vnode 信息并开始消费。
如下图所示，c1 表示消费者 1，c2 表示消费者 2，g1 表示消费组 1。起初 g1 中只有 c1 消费数据，c1 发送订阅信息到 mnode，mnode 把数据所在的所有 4 个vnode 分配给 c1。当 c2 增加到 g1 后，c2 将订阅信息发送给 mnode，mnode 检测到这个 g1 需要重新分配，就会启动 rebalance 过程，随后 c2 分得其中两个 vnode 消费。分配信息还会通过 mnode 发送给 vnode，同时 c1 和 c2 也会获取自己消费的 vnode 信息并开始消费。
![再平衡过程](./rebalance.png)
![rebalance 过程](./rebalance.png)
再平衡计时器每 2s 检测一次是否需要再平衡。在再平衡过程中，如果消费者获取的状态是 not ready，则不能进行消费。只有再平衡正常结束后，消费者获取分配 vnode 的offset 后才可正常消费，否则消费者会重试指定次数后报错。
## 消费者状态处理
消费者的状态转换过程如下图所示。初始状态下，刚发起订阅的消费者处于modify 状态，此时客户端获取的消费者状态为 not ready，表明消费者尚未准备好进行数据消费。一旦 mnode 在再平衡计时器中检测到处于 modify 状态的消费者，便会启动再平衡过程。再平衡成功后，消费者的状态将转变为 ready，表示消费者已准备就绪。随后，当消费者通过定时的查询 endpoint 消息获取自身的 ready 状态以及分配的 vnode 列表后，便能正式开始消费数据。
消费者的状态转换过程如下图所示。初始状态下，刚发起订阅的消费者处于 rebalancing 状态，表明消费者尚未准备好进行数据消费。一旦 mnode 检测到处于 rebalancing 状态的消费者，便会启动 rebalance 过程，成功后，消费者的状态将转变为 ready，表示消费者已准备就绪。随后，当消费者通过定时的查询 endpoint 消息获取自身的 ready 状态以及分配的 vnode 列表后，便能正式开始消费数据。
![消费者状态转换](./statetransition.png)
若消费者的心跳丢失时间超过 12s，经过再平衡过程，其状态将被更新为 lost，表明消费者被认为已离线。如果心跳丢失时间超过 1 天，消费者的状态将进一步转变为 clear，此时消费者将被系统删除。然而，如果在上述过程中消费者重新发送心跳信号，其状态将恢复为 modify，并重新进入下一轮的再平衡过程。
若消费者的心跳丢失时间超过 12s，经过 rebalance 过程，其状态将被更新为 clear，然后消费者将被系统删除。
当消费者主动退出时，会发送 unsubscribe 消息。该消息将清除消费者订阅的所有主题，并将消费者的状态设置为 modify。随后，在再平衡过程中，一旦消费者的状态变为ready 且无订阅的主题，mnode 将在再平衡的计时器中检测到此状态变化，并据此删除该消费者。这一系列措施确保了消费者退出的有序性和系统的稳定性。
当消费者主动退出时，会发送 unsubscribe 消息。该消息将清除消费者订阅的所有主题，并将消费者的状态设置为 rebalancing。随后，检测到处于 rebalancing 状态的消费者，便会启动 rebalance 过程，成功后，其状态将被更新为 clear，然后消费者将被系统删除。这一系列措施确保了消费者退出的有序性和系统的稳定性。
## 消费数据

BIN
docs/zh/26-tdinternal/consumer.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 118 KiB

After

Width:  |  Height:  |  Size: 48 KiB

BIN
docs/zh/26-tdinternal/offset.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 46 KiB

After

Width:  |  Height:  |  Size: 15 KiB

BIN
docs/zh/26-tdinternal/rebalance.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 232 KiB

After

Width:  |  Height:  |  Size: 282 KiB

BIN
docs/zh/26-tdinternal/statetransition.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 72 KiB

After

Width:  |  Height:  |  Size: 40 KiB

BIN
docs/zh/26-tdinternal/topic.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 56 KiB

After

Width:  |  Height:  |  Size: 18 KiB

BIN
docs/zh/26-tdinternal/topicarch.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 215 KiB

After

Width:  |  Height:  |  Size: 103 KiB

4
include/common/tmsg.h
View File

@ -2817,8 +2817,8 @@ enum {
TOPIC_SUB_TYPE__COLUMN,
};
#define DEFAULT_MAX_POLL_INTERVAL 3000000
#define DEFAULT_SESSION_TIMEOUT 12000
#define DEFAULT_MAX_POLL_INTERVAL 300000
#define DEFAULT_SESSION_TIMEOUT 12000
typedef struct {
char name[TSDB_TOPIC_FNAME_LEN]; // accout.topic

2
include/common/tvariant.h
View File

@ -49,7 +49,7 @@ int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value);
*/
int32_t toIntegerPure(const char *z, int32_t n, int32_t base, int64_t *value);
void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type);
int32_t taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type);
void taosVariantDestroy(SVariant *pV);

4
include/dnode/mnode/mnode.h
View File

@ -119,9 +119,9 @@ void mndPostProcessQueryMsg(SRpcMsg *pMsg);
*/
void mndGenerateMachineCode();
void mndDumpSdb();
int32_t mndDumpSdb();
void mndDeleteTrans();
int32_t mndDeleteTrans();
#ifdef __cplusplus
}

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

@ -49,6 +49,7 @@ typedef struct {
} SAuditRecord;
int32_t auditInit(const SAuditCfg *pCfg);
void auditSetDnodeId(int32_t dnodeId);
void auditCleanup();
int32_t auditSend(SJson *pJson);
void auditRecord(SRpcMsg *pReq, int64_t clusterId, char *operation, char *target1, char *target2,

2
include/libs/executor/storageapi.h
View File

@ -197,7 +197,7 @@ typedef struct SStoreCacheReader {
SArray *pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks);
void (*closeReader)(void *pReader);
int32_t (*retrieveRows)(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, const int32_t *dstSlotIds,
SArray *pTableUidList);
SArray *pTableUidList, bool* pGotAllRows);
int32_t (*reuseReader)(void *pReader, void *pTableIdList, int32_t numOfTables);
} SStoreCacheReader;

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

@ -218,6 +218,7 @@ typedef struct {
} SDmNotifyHandle;
int32_t monInit(const SMonCfg *pCfg);
void monSetDnodeId(int32_t dnodeId);
void monInitVnode();
void monCleanup();
void monRecordLog(int64_t ts, ELogLevel level, const char *content);

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

@ -173,6 +173,7 @@ int32_t nodesNodeToSQL(SNode* pNode, char* buf, int32_t bufSize, int32_t* len);
char* nodesGetNameFromColumnNode(SNode* pNode);
int32_t nodesGetOutputNumFromSlotList(SNodeList* pSlots);
void nodesSortList(SNodeList** pList, int32_t (*)(SNode* pNode1, SNode* pNode2));
void destroyFuncParam(void* pFuncStruct);
#ifdef __cplusplus
}

1
include/libs/nodes/querynodes.h
View File

@ -57,7 +57,6 @@ typedef struct SExprNode {
char aliasName[TSDB_COL_NAME_LEN];
char userAlias[TSDB_COL_NAME_LEN];
SArray* pAssociation;
bool orderAlias;
bool asAlias;
bool asParam;
bool asPosition;

4
include/libs/transport/thttp.h
View File

@ -28,9 +28,11 @@ typedef enum { HTTP_GZIP, HTTP_FLAT } EHttpCompFlag;
int32_t taosSendHttpReport(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag);
int32_t taosSendHttpReportWithQID(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag, const char* qid);
int64_t taosInitHttpChan();
int32_t taosSendHttpReportByChan(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen,
EHttpCompFlag flag, int64_t chanId);
EHttpCompFlag flag, int64_t chanId, const char* qid);
void taosDestroyHttpChan(int64_t chanId);
#ifdef __cplusplus

4
include/libs/wal/wal.h
View File

@ -112,8 +112,8 @@ typedef struct SWal {
int64_t totSize;
int64_t lastRollSeq;
// ctl
int64_t refId;
TdThreadMutex mutex;
int64_t refId;
TdThreadRwlock mutex;
// ref
SHashObj *pRefHash; // refId -> SWalRef
// path

1
include/os/osSocket.h
View File

@ -165,7 +165,6 @@ int32_t taosGetFqdn(char *);
void tinet_ntoa(char *ipstr, uint32_t ip);
uint32_t ip2uint(const char *const ip_addr);
int32_t taosIgnSIGPIPE();
uint32_t taosInetAddr(const char *ipAddr);
const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len);
uint64_t taosHton64(uint64_t val);

2
include/os/osSysinfo.h
View File

@ -80,7 +80,7 @@ typedef struct {
SysNameInfo taosGetSysNameInfo();
bool taosCheckCurrentInDll();
int taosGetlocalhostname(char *hostname, size_t maxLen);
int32_t taosGetlocalhostname(char *hostname, size_t maxLen);
#ifdef __cplusplus
}

5
include/os/osSystem.h
View File

@ -81,8 +81,11 @@ int32_t taosResetTerminalMode();
unw_get_reg(&cursor, UNW_REG_IP, &pc); \
fname[0] = '\0'; \
(void)unw_get_proc_name(&cursor, fname, sizeof(fname), &offset); \
size += 1; \
array[size] = (char *)taosMemoryMalloc(sizeof(char) * STACKSIZE + 1); \
if(NULL == array[size]) { \
break; \
} \
size += 1; \
snprintf(array[size], STACKSIZE, "0x%lx : (%s+0x%lx) [0x%lx]\n", (long)pc, fname, (long)offset, (long)pc); \
} \
if (ignoreNum < size && size > 0) { \

2
include/util/tpagedbuf.h
View File

@ -149,7 +149,7 @@ void setBufPageDirty(void* pPage, bool dirty);
* Set the compress/ no-compress flag for paged buffer, when flushing data in disk.
* @param pBuf
*/
void setBufPageCompressOnDisk(SDiskbasedBuf* pBuf, bool comp);
int32_t setBufPageCompressOnDisk(SDiskbasedBuf* pBuf, bool comp);
/**
* Set the pageId page buffer is not need

22
include/util/tutil.h
View File

@ -73,14 +73,14 @@ static FORCE_INLINE void taosEncryptPass_c(uint8_t *inBuf, size_t len, char *tar
char buf[TSDB_PASSWORD_LEN + 1];
buf[TSDB_PASSWORD_LEN] = 0;
(void)sprintf(buf, "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x", context.digest[0], context.digest[1],
context.digest[2], context.digest[3], context.digest[4], context.digest[5], context.digest[6],
context.digest[7], context.digest[8], context.digest[9], context.digest[10], context.digest[11],
context.digest[12], context.digest[13], context.digest[14], context.digest[15]);
(void)sprintf(buf, "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x", context.digest[0],
context.digest[1], context.digest[2], context.digest[3], context.digest[4], context.digest[5],
context.digest[6], context.digest[7], context.digest[8], context.digest[9], context.digest[10],
context.digest[11], context.digest[12], context.digest[13], context.digest[14], context.digest[15]);
(void)memcpy(target, buf, TSDB_PASSWORD_LEN);
}
static FORCE_INLINE int32_t taosHashBinary(char* pBuf, int32_t len) {
static FORCE_INLINE int32_t taosHashBinary(char *pBuf, int32_t len) {
uint64_t hashVal = MurmurHash3_64(pBuf, len);
return sprintf(pBuf, "%" PRIu64, hashVal);
}
@ -161,8 +161,7 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
#define TCONTAINER_OF(ptr, type, member) ((type *)((char *)(ptr)-offsetof(type, member)))
#define TAOS_GET_TERRNO(code) \
(terrno == 0 ? code : terrno)
#define TAOS_GET_TERRNO(code) (terrno == 0 ? code : terrno)
#define TAOS_RETURN(CODE) \
do { \
@ -177,6 +176,15 @@ static FORCE_INLINE int32_t taosGetTbHashVal(const char *tbname, int32_t tblen,
} \
} while (0)
#define TAOS_CHECK_RETURN_WITH_FREE(CMD, PTR) \
do { \
int32_t __c = (CMD); \
if (__c != TSDB_CODE_SUCCESS) { \
taosMemoryFree(PTR); \
TAOS_RETURN(__c); \
} \
} while (0)
#define TAOS_CHECK_GOTO(CMD, LINO, LABEL) \
do { \
code = (CMD); \

11
include/util/tuuid.h
View File

@ -37,3 +37,14 @@ int32_t tGenIdPI32(void);
* @return
*/
int64_t tGenIdPI64(void);
/**
* Generate an qid
*+------------+-----+-----------+---------------+
*| nodeid| 0| serial number | 0 |
*+------------+-----+-----------+---------------+
*| 8bit | 16bit|32bit |8bit |
*+------------+-----+-----------+---------------+
* @return
*/
int64_t tGenQid64(int8_t dnodeId);

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

@ -552,6 +552,9 @@ int32_t createRequest(uint64_t connId, int32_t type, int64_t reqid, SRequestObj
(*pRequest)->allocatorRefId = -1;
(*pRequest)->pDb = getDbOfConnection(pTscObj);
if (NULL == (*pRequest)->pDb) {
TSC_ERR_JRET(terrno);
}
(*pRequest)->pTscObj = pTscObj;
(*pRequest)->inCallback = false;
(*pRequest)->msgBuf = taosMemoryCalloc(1, ERROR_MSG_BUF_DEFAULT_SIZE);

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

@ -1359,6 +1359,14 @@ static void *hbThreadFunc(void *param) {
pInfo->msgInfo.len = tlen;
pInfo->msgType = TDMT_MND_HEARTBEAT;
pInfo->param = taosMemoryMalloc(sizeof(int32_t));
if (pInfo->param == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tFreeClientHbBatchReq(pReq);
// hbClearReqInfo(pAppHbMgr);
taosMemoryFree(buf);
taosMemoryFree(pInfo);
break;
}
*(int32_t *)pInfo->param = i;
pInfo->paramFreeFp = taosMemoryFree;
pInfo->requestId = generateRequestId();

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

@ -37,7 +37,6 @@ void setQueryRequest(int64_t rId) {
pReq->isQuery = true;
(void)releaseRequest(rId);
}
}
static bool stringLengthCheck(const char* str, size_t maxsize) {
@ -1291,7 +1290,7 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue
}
if (TSDB_CODE_SUCCESS != code) {
pRequest->code = terrno;
pRequest->code = code;
}
if (res) {
@ -1586,6 +1585,8 @@ int32_t taosConnectImpl(const char* user, const char* auth, const char* db, __ta
pRequest->sqlstr = taosStrdup("taos_connect");
if (pRequest->sqlstr) {
pRequest->sqlLen = strlen(pRequest->sqlstr);
} else {
return terrno;
}
SMsgSendInfo* body = NULL;
@ -1649,6 +1650,9 @@ static int32_t buildConnectMsg(SRequestObj* pRequest, SMsgSendInfo** pMsgSendInf
char* db = getDbOfConnection(pObj);
if (db != NULL) {
tstrncpy(connectReq.db, db, sizeof(connectReq.db));
} else if (terrno) {
taosMemoryFree(*pMsgSendInfo);
return terrno;
}
taosMemoryFreeClear(db);
@ -2043,7 +2047,7 @@ static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int32_t numOfCols, int32_t* colLength) {
int32_t idx = -1;
iconv_t conv = taosAcquireConv(&idx, C2M);
if (!conv) return TSDB_CODE_TSC_INTERNAL_ERROR;
if (conv == (iconv_t)-1) return TSDB_CODE_TSC_INTERNAL_ERROR;
for (int32_t i = 0; i < numOfCols; ++i) {
int32_t type = pResultInfo->fields[i].type;
@ -2240,6 +2244,10 @@ static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int
} else if (tTagIsJson(data)) {
char* jsonString = NULL;
parseTagDatatoJson(data, &jsonString);
if (jsonString == NULL) {
tscError("doConvertJson error: parseTagDatatoJson failed");
return terrno;
}
STR_TO_VARSTR(dst, jsonString);
taosMemoryFree(jsonString);
} else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) { // value -> "value"
@ -2392,11 +2400,16 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
}
char* getDbOfConnection(STscObj* pObj) {
terrno = TSDB_CODE_SUCCESS;
char* p = NULL;
(void)taosThreadMutexLock(&pObj->mutex);
size_t len = strlen(pObj->db);
if (len > 0) {
p = strndup(pObj->db, tListLen(pObj->db));
if (p == NULL) {
tscError("failed to strndup db name");
terrno = TSDB_CODE_OUT_OF_MEMORY;
}
}
(void)taosThreadMutexUnlock(&pObj->mutex);
@ -2503,10 +2516,10 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
void* clientRpc = NULL;
SServerStatusRsp statusRsp = {0};
SEpSet epSet = {.inUse = 0, .numOfEps = 1};
SRpcMsg rpcMsg = {.info.ahandle = (void*)0x9526, .msgType = TDMT_DND_SERVER_STATUS};
SRpcMsg rpcRsp = {0};
SRpcInit rpcInit = {0};
char pass[TSDB_PASSWORD_LEN + 1] = {0};
SRpcMsg rpcMsg = {.info.ahandle = (void*)0x9527, .info.notFreeAhandle = 1, .msgType = TDMT_DND_SERVER_STATUS};
SRpcMsg rpcRsp = {0};
SRpcInit rpcInit = {0};
char pass[TSDB_PASSWORD_LEN + 1] = {0};
rpcInit.label = "CHK";
rpcInit.numOfThreads = 1;
@ -3018,7 +3031,8 @@ void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param
void doRequestCallback(SRequestObj* pRequest, int32_t code) {
pRequest->inCallback = true;
int64_t this = pRequest->self;
if (tsQueryTbNotExistAsEmpty && TD_RES_QUERY(&pRequest->resType) && pRequest->isQuery && (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TDB_TABLE_NOT_EXIST)) {
if (tsQueryTbNotExistAsEmpty && TD_RES_QUERY(&pRequest->resType) && pRequest->isQuery &&
(code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TDB_TABLE_NOT_EXIST)) {
code = TSDB_CODE_SUCCESS;
pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
}

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

@ -216,9 +216,7 @@ static void reportSendProcess(void* param, void* tmrId) {
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
generateClusterReport(pMonitor->registry, pInst->pTransporter, &ep);
bool reset = taosTmrReset(reportSendProcess, pInst->monitorParas.tsMonitorInterval * 1000, param, monitorTimer, &tmrId);
if (!reset){
tscError("failed to reset timer, pMonitor:%p", pMonitor);
}
tscDebug("reset timer, pMonitor:%p, %d", pMonitor, reset);
taosRUnLockLatch(&monitorLock);
}
@ -736,6 +734,16 @@ static void monitorSendAllSlowLogFromTempDir(int64_t clusterId) {
continue;
}
char* tmp = taosStrdup(filename);
if (tmp == NULL) {
tscError("failed to dup string:%s since %s", filename, terrstr());
if (taosUnLockFile(pFile) != 0) {
tscError("failed to unlock file:%s, terrno:%d", filename, terrno);
}
if (taosCloseFile(&(pFile)) != 0) {
tscError("failed to close file:%s, terrno:%d", filename, terrno);
}
continue;
}
monitorSendSlowLogAtBeginning(clusterId, &tmp, pFile, 0);
taosMemoryFree(tmp);
}

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

@ -417,6 +417,10 @@ static void buildChildElement(cJSON* json, SVCreateTbReq* pCreateReq) {
}
char* pJson = NULL;
parseTagDatatoJson(pTag, &pJson);
if(pJson == NULL) {
uError("parseTagDatatoJson failed, pJson == NULL");
goto end;
}
cJSON* tag = cJSON_CreateObject();
RAW_NULL_CHECK(tag);
STagVal* pTagVal = taosArrayGet(pTagVals, 0);
@ -727,6 +731,10 @@ static void processAlterTable(SMqMetaRsp* metaRsp, cJSON** pJson) {
goto end;
}
parseTagDatatoJson(vAlterTbReq.pTagVal, &buf);
if(buf == NULL) {
uError("parseTagDatatoJson failed, buf == NULL");
goto end;
}
} else {
if (vAlterTbReq.tagType == TSDB_DATA_TYPE_VARBINARY) {
buf = taosMemoryCalloc(vAlterTbReq.nTagVal * 2 + 2 + 3, 1);

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

@ -1939,6 +1939,10 @@ int32_t smlClearForRerun(SSmlHandle *info) {
return TSDB_CODE_SML_INVALID_DATA;
}
info->lines = (SSmlLineInfo *)taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));
if (unlikely(info->lines == NULL)) {
uError("SML:0x%" PRIx64 " info->lines == NULL", info->id);
return terrno;
}
}
(void)memset(&info->preLine, 0, sizeof(SSmlLineInfo));
@ -1971,10 +1975,14 @@ static bool getLine(SSmlHandle *info, char *lines[], char **rawLine, char *rawLi
if (*rawLine != NULL && (uDebugFlag & DEBUG_DEBUG)) {
char *print = taosMemoryCalloc(*len + 1, 1);
(void)memcpy(print, *tmp, *len);
uDebug("SML:0x%" PRIx64 " smlParseLine is raw, numLines:%d, protocol:%d, len:%d, data:%s", info->id, numLines,
info->protocol, *len, print);
taosMemoryFree(print);
if (print != NULL){
(void)memcpy(print, *tmp, *len);
uDebug("SML:0x%" PRIx64 " smlParseLine is raw, numLines:%d, protocol:%d, len:%d, data:%s", info->id, numLines,
info->protocol, *len, print);
taosMemoryFree(print);
} else{
uError("SML:0x%" PRIx64 " smlParseLine taosMemoryCalloc failed", info->id);
}
} else {
uDebug("SML:0x%" PRIx64 " smlParseLine is not numLines:%d, protocol:%d, len:%d, data:%s", info->id, numLines,
info->protocol, *len, *tmp);

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

@ -924,6 +924,9 @@ int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
}
pStmt->sql.sqlStr = strndup(sql, length);
if (!pStmt->sql.sqlStr) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pStmt->sql.sqlLen = length;
pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;

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

@ -194,6 +194,12 @@ static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void
pStmt->bInfo.tbSuid = pTableMeta->suid;
pStmt->bInfo.tbVgId = pTableMeta->vgId;
pStmt->bInfo.tbType = pTableMeta->tableType;
if (!pStmt->bInfo.tagsCached) {
qDestroyBoundColInfo(pStmt->bInfo.boundTags);
taosMemoryFreeClear(pStmt->bInfo.boundTags);
}
pStmt->bInfo.boundTags = tags;
pStmt->bInfo.tagsCached = false;
tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
@ -863,6 +869,9 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
}
pStmt->sql.sqlStr = strndup(sql, length);
if (!pStmt->sql.sqlStr) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pStmt->sql.sqlLen = length;
pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
@ -985,10 +994,6 @@ int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags) {
return TSDB_CODE_SUCCESS;
}
if (pStmt->bInfo.inExecCache) {
return TSDB_CODE_SUCCESS;
}
STableDataCxt** pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
@ -996,6 +1001,10 @@ int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags) {
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
}
if (pStmt->bInfo.inExecCache && !pStmt->sql.autoCreateTbl) {
return TSDB_CODE_SUCCESS;
}
tscDebug("start to bind stmt tag values");
STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,

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

@ -300,6 +300,7 @@ void tmq_conf_destroy(tmq_conf_t* conf) {
tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value) {
int32_t code = 0;
if (conf == NULL || key == NULL || value == NULL) {
tscError("tmq_conf_set null, conf:%p key:%p value:%p", conf, key, value);
return TMQ_CONF_INVALID;
}
if (strcasecmp(key, "group.id") == 0) {
@ -320,6 +321,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
conf->autoCommit = false;
return TMQ_CONF_OK;
} else {
tscError("invalid value for enable.auto.commit: %s", value);
return TMQ_CONF_INVALID;
}
}
@ -328,6 +330,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 0 || code != 0) {
tscError("invalid value for auto.commit.interval.ms: %s", value);
return TMQ_CONF_INVALID;
}
conf->autoCommitInterval = (tmp > INT32_MAX ? INT32_MAX : tmp);
@ -338,6 +341,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 6000 || tmp > 1800000 || code != 0) {
tscError("invalid value for session.timeout.ms: %s", value);
return TMQ_CONF_INVALID;
}
conf->sessionTimeoutMs = tmp;
@ -348,6 +352,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp < 1000 || tmp >= conf->sessionTimeoutMs || code != 0) {
tscError("invalid value for heartbeat.interval.ms: %s", value);
return TMQ_CONF_INVALID;
}
conf->heartBeatIntervalMs = tmp;
@ -358,6 +363,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
int32_t tmp;
code = taosStr2int32(value, &tmp);
if (tmp < 1000 || code != 0) {
tscError("invalid value for max.poll.interval.ms: %s", value);
return TMQ_CONF_INVALID;
}
conf->maxPollIntervalMs = tmp;
@ -375,6 +381,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
conf->resetOffset = TMQ_OFFSET__RESET_LATEST;
return TMQ_CONF_OK;
} else {
tscError("invalid value for auto.offset.reset: %s", value);
return TMQ_CONF_INVALID;
}
}
@ -387,6 +394,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
conf->withTbName = false;
return TMQ_CONF_OK;
} else {
tscError("invalid value for msg.with.table.name: %s", value);
return TMQ_CONF_INVALID;
}
}
@ -399,22 +407,38 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
conf->snapEnable = false;
return TMQ_CONF_OK;
} else {
tscError("invalid value for experimental.snapshot.enable: %s", value);
return TMQ_CONF_INVALID;
}
}
if (strcasecmp(key, "td.connect.ip") == 0) {
conf->ip = taosStrdup(value);
void *tmp = taosStrdup(value);
if (tmp == NULL) {
tscError("tmq_conf_set out of memory:%d", terrno);
return TMQ_CONF_INVALID;
}
conf->ip = tmp;
return TMQ_CONF_OK;
}
if (strcasecmp(key, "td.connect.user") == 0) {
conf->user = taosStrdup(value);
void *tmp = taosStrdup(value);
if (tmp == NULL) {
tscError("tmq_conf_set out of memory:%d", terrno);
return TMQ_CONF_INVALID;
}
conf->user = tmp;
return TMQ_CONF_OK;
}
if (strcasecmp(key, "td.connect.pass") == 0) {
conf->pass = taosStrdup(value);
void *tmp = taosStrdup(value);
if (tmp == NULL) {
tscError("tmq_conf_set out of memory:%d", terrno);
return TMQ_CONF_INVALID;
}
conf->pass = tmp;
return TMQ_CONF_OK;
}
@ -422,6 +446,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
int64_t tmp;
code = taosStr2int64(value, &tmp);
if (tmp <= 0 || tmp > 65535 || code != 0) {
tscError("invalid value for td.connect.port: %s", value);
return TMQ_CONF_INVALID;
}
@ -437,6 +462,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
conf->replayEnable = false;
return TMQ_CONF_OK;
} else {
tscError("invalid value for enable.replay: %s", value);
return TMQ_CONF_INVALID;
}
}
@ -458,6 +484,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
return TMQ_CONF_OK;
}
tscError("unknown key: %s", key);
return TMQ_CONF_UNKNOWN;
}
@ -468,7 +495,11 @@ int32_t tmq_list_append(tmq_list_t* list, const char* src) {
SArray* container = &list->container;
if (src == NULL || src[0] == 0) return TSDB_CODE_INVALID_PARA;
char* topic = taosStrdup(src);
if (taosArrayPush(container, &topic) == NULL) return TSDB_CODE_INVALID_PARA;
if (topic == NULL) return terrno;
if (taosArrayPush(container, &topic) == NULL) {
taosMemoryFree(topic);
return terrno;
}
return 0;
}
@ -947,13 +978,13 @@ void tmqSendHbReq(void* param, void* tmrId) {
int32_t tlen = tSerializeSMqHbReq(NULL, 0, &req);
if (tlen < 0) {
tscError("tSerializeSMqHbReq failed");
tscError("tSerializeSMqHbReq failed, size:%d", tlen);
goto OVER;
}
void* pReq = taosMemoryCalloc(1, tlen);
if (tlen < 0) {
tscError("failed to malloc MqHbReq msg, size:%d", tlen);
if (pReq == NULL) {
tscError("failed to malloc MqHbReq msg, code:%d", terrno);
goto OVER;
}
@ -989,9 +1020,7 @@ OVER:
tDestroySMqHbReq(&req);
if (tmrId != NULL) {
bool ret = taosTmrReset(tmqSendHbReq, tmq->heartBeatIntervalMs, param, tmqMgmt.timer, &tmq->hbLiveTimer);
if (!ret){
tscError("failed to reset timer fo tmq hb");
}
tscDebug("reset timer fo tmq hb:%d", ret);
}
int32_t ret = taosReleaseRef(tmqMgmt.rsetId, refId);
if (ret != 0){
@ -1033,9 +1062,7 @@ void tmqHandleAllDelayedTask(tmq_t* pTmq) {
tscDebug("consumer:0x%" PRIx64 " retrieve ep from mnode in 1s", pTmq->consumerId);
bool ret = taosTmrReset(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(pTmq->refId), tmqMgmt.timer,
&pTmq->epTimer);
if (!ret){
tscError("failed to reset timer fo tmq ask ep");
}
tscDebug("reset timer fo tmq ask ep:%d", ret);
} else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
tmq_commit_cb* pCallbackFn = pTmq->commitCb ? pTmq->commitCb : defaultCommitCbFn;
asyncCommitAllOffsets(pTmq, pCallbackFn, pTmq->commitCbUserParam);
@ -1043,9 +1070,7 @@ void tmqHandleAllDelayedTask(tmq_t* pTmq) {
pTmq->autoCommitInterval / 1000.0);
bool ret = taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer,
&pTmq->commitTimer);
if (!ret){
tscError("failed to reset timer fo commit");
}
tscDebug("reset timer fo commit:%d", ret);
} else {
tscError("consumer:0x%" PRIx64 " invalid task type:%d", pTmq->consumerId, *pTaskType);
}
@ -3514,6 +3539,10 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a
}
(void)taosThreadMutexInit(&pCommon->mutex, 0);
pCommon->pTopicName = taosStrdup(pTopic->topicName);
if (pCommon->pTopicName == NULL) {
code = terrno;
goto end;
}
pCommon->consumerId = tmq->consumerId;
for (int32_t i = 0; i < (*numOfAssignment); ++i) {

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

@ -2620,11 +2620,15 @@ int32_t dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf
}
len += snprintf(dumpBuf + len, size - len, "%s |end\n", flag);
*pDataBuf = dumpBuf;
dumpBuf = NULL;
_exit:
if (dumpBuf) {
taosMemoryFree(dumpBuf);
if (code == TSDB_CODE_SUCCESS) {
*pDataBuf = dumpBuf;
dumpBuf = NULL;
} else {
uError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
if (dumpBuf) {
taosMemoryFree(dumpBuf);
}
}
return code;
}

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

@ -1092,6 +1092,9 @@ int32_t taosSetSlowLogScope(char *pScopeStr, int32_t *pScope) {
while((scope = strsep(&pScopeStr, "|")) != NULL){
taosMemoryFreeClear(tmp);
tmp = taosStrdup(scope);
if (tmp == NULL) {
TAOS_RETURN(terrno);
}
(void)strtrim(tmp);
if (0 == strcasecmp(tmp, "all")) {
slowScope |= SLOW_LOG_TYPE_ALL;

2
source/common/src/tname.c
View File

@ -143,7 +143,7 @@ bool tNameIsValid(const SName* name) {
SName* tNameDup(const SName* name) {
SName* p = taosMemoryMalloc(sizeof(SName));
memcpy(p, name, sizeof(SName));
if (p) TAOS_MEMCPY(p, name, sizeof(SName));
return p;
}

4
source/common/src/tvariant.c
View File

@ -368,7 +368,7 @@ int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value) {
* @param len
* @param type
*/
void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type) {
int32_t taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type) {
switch (type) {
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT: {
@ -426,6 +426,7 @@ void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uin
size_t lenInwchar = len / TSDB_NCHAR_SIZE;
pVar->ucs4 = taosMemoryCalloc(1, (lenInwchar + 1) * TSDB_NCHAR_SIZE);
if(!pVar->ucs4) return terrno;
(void)memcpy(pVar->ucs4, pz, lenInwchar * TSDB_NCHAR_SIZE);
pVar->nLen = (int32_t)len;
@ -446,6 +447,7 @@ void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uin
}
pVar->nType = type;
return 0;
}
void taosVariantDestroy(SVariant *pVar) {

6
source/dnode/mgmt/exe/dmMain.c
View File

@ -433,7 +433,8 @@ int mainWindows(int argc, char **argv) {
}
if (global.dumpSdb) {
mndDumpSdb();
int32_t code = 0;
TAOS_CHECK_RETURN(mndDumpSdb());
taosCleanupCfg();
taosCloseLog();
taosCleanupArgs();
@ -442,6 +443,7 @@ int mainWindows(int argc, char **argv) {
}
if (global.deleteTrans) {
int32_t code = 0;
TdFilePtr pFile;
if ((code = dmCheckRunning(tsDataDir, &pFile)) != 0) {
printf("failed to generate encrypt code since taosd is running, please stop it first, reason:%s",
@ -449,7 +451,7 @@ int mainWindows(int argc, char **argv) {
return code;
}
mndDeleteTrans();
TAOS_CHECK_RETURN(mndDeleteTrans());
taosCleanupCfg();
taosCloseLog();
taosCleanupArgs();

9
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
View File

@ -14,7 +14,9 @@
*/
#define _DEFAULT_SOURCE
#include "audit.h"
#include "dmInt.h"
#include "monitor.h"
#include "systable.h"
#include "tchecksum.h"
@ -27,6 +29,8 @@ static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) {
(void)taosThreadRwlockWrlock(&pMgmt->pData->lock);
pMgmt->pData->dnodeId = pCfg->dnodeId;
pMgmt->pData->clusterId = pCfg->clusterId;
monSetDnodeId(pCfg->dnodeId);
auditSetDnodeId(pCfg->dnodeId);
code = dmWriteEps(pMgmt->pData);
if (code != 0) {
dInfo("failed to set local info, dnodeId:%d clusterId:%" PRId64 " reason:%s", pCfg->dnodeId, pCfg->clusterId,
@ -65,6 +69,7 @@ static void dmMayShouldUpdateIpWhiteList(SDnodeMgmt *pMgmt, int64_t ver) {
.contLen = contLen,
.msgType = TDMT_MND_RETRIEVE_IP_WHITE,
.info.ahandle = (void *)0x9527,
.info.notFreeAhandle = 1,
.info.refId = 0,
.info.noResp = 0,
.info.handle = 0};
@ -181,6 +186,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
.contLen = contLen,
.msgType = TDMT_MND_STATUS,
.info.ahandle = (void *)0x9527,
.info.notFreeAhandle = 1,
.info.refId = 0,
.info.noResp = 0,
.info.handle = 0};
@ -230,6 +236,7 @@ void dmSendNotifyReq(SDnodeMgmt *pMgmt, SNotifyReq *pReq) {
.contLen = contLen,
.msgType = TDMT_MND_NOTIFY,
.info.ahandle = (void *)0x9527,
.info.notFreeAhandle = 1,
.info.refId = 0,
.info.noResp = 1,
.info.handle = 0};
@ -479,7 +486,7 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t len = blockEncode(pBlock, pStart, numOfCols);
if(len < 0) {
if (len < 0) {
dError("failed to retrieve data since %s", tstrerror(code));
blockDataDestroy(pBlock);
rpcFreeCont(pRsp);

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

@ -360,6 +360,12 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
SVnodeThread *threads = taosMemoryCalloc(threadNum, sizeof(SVnodeThread));
if (threads == NULL) {
dError("failed to allocate memory for threads since %s", terrstr());
taosMemoryFree(pCfgs);
return terrno;
}
for (int32_t t = 0; t < threadNum; ++t) {
threads[t].threadIndex = t;
threads[t].pMgmt = pMgmt;

6
source/dnode/mgmt/node_mgmt/src/dmNodes.c
View File

@ -15,6 +15,9 @@
#define _DEFAULT_SOURCE
#include "dmMgmt.h"
#include "dmUtil.h"
#include "monitor.h"
#include "audit.h"
int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
int32_t code = 0;
@ -98,6 +101,9 @@ static int32_t dmOpenNodes(SDnode *pDnode) {
}
}
auditSetDnodeId(dmGetDnodeId(&pDnode->data));
monSetDnodeId(dmGetDnodeId(&pDnode->data));
dmSetStatus(pDnode, DND_STAT_RUNNING);
return 0;
}

11
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -24,6 +24,12 @@ static inline void dmBuildMnodeRedirectRsp(SDnode *pDnode, SRpcMsg *pMsg) {
SEpSet epSet = {0};
dmGetMnodeEpSetForRedirect(&pDnode->data, pMsg, &epSet);
if (epSet.numOfEps <= 1) {
pMsg->pCont = NULL;
pMsg->code = TSDB_CODE_MNODE_NOT_FOUND;
return;
}
int32_t contLen = tSerializeSEpSet(NULL, 0, &epSet);
pMsg->pCont = rpcMallocCont(contLen);
if (pMsg->pCont == NULL) {
@ -109,7 +115,8 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
int32_t svrVer = 0;
(void)taosVersionStrToInt(version, &svrVer);
if ((code = taosCheckVersionCompatible(pRpc->info.cliVer, svrVer, 3)) != 0) {
dError("Version not compatible, cli ver: %d, svr ver: %d, ip:0x%x", pRpc->info.cliVer, svrVer, pRpc->info.conn.clientIp);
dError("Version not compatible, cli ver: %d, svr ver: %d, ip:0x%x", pRpc->info.cliVer, svrVer,
pRpc->info.conn.clientIp);
goto _OVER;
}
@ -437,7 +444,7 @@ int32_t dmInitStatusClient(SDnode *pDnode) {
pTrans->statusRpc = rpcOpen(&rpcInit);
if (pTrans->statusRpc == NULL) {
dError("failed to init dnode rpc status client since %s", tstrerror(terrno));
dError("failed to init dnode rpc status client since %s", tstrerror(terrno));
return terrno;
}

1
source/dnode/mgmt/node_util/inc/dmUtil.h
View File

@ -217,6 +217,7 @@ int32_t dmInitDndInfo(SDnodeData *pData);
// dmEps.c
int32_t dmGetDnodeSize(SDnodeData *pData);
int32_t dmGetDnodeId(SDnodeData *pData);
int32_t dmReadEps(SDnodeData *pData);
int32_t dmWriteEps(SDnodeData *pData);
void dmUpdateEps(SDnodeData *pData, SArray *pDnodeEps);

2
source/dnode/mgmt/node_util/src/dmUtil.c
View File

@ -88,3 +88,5 @@ void dmGetMonitorSystemInfo(SMonSysInfo *pInfo) {
}
return;
}
int32_t dmGetDnodeId(SDnodeData *pData) { return pData->dnodeId; }

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

@ -81,7 +81,7 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
code = terrno;
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
mInfo("acct:%s, will be created when deploying, raw:%p", acctObj.acct, pRaw);

52
source/dnode/mnode/impl/src/mndArbGroup.c
View File

@ -256,7 +256,9 @@ static int32_t mndArbGroupActionUpdate(SSdb *pSdb, SArbGroup *pOld, SArbGroup *p
_OVER:
(void)taosThreadMutexUnlock(&pOld->mutex);
(void)taosHashRemove(arbUpdateHash, &pOld->vgId, sizeof(int32_t));
if (taosHashRemove(arbUpdateHash, &pOld->vgId, sizeof(int32_t)) != 0) {
mError("arbgroup:%d, failed to remove from arbUpdateHash", pOld->vgId);
}
return 0;
}
@ -451,7 +453,7 @@ static int32_t mndProcessArbHbTimer(SRpcMsg *pReq) {
int64_t mndTerm = mndGetTerm(pMnode);
if (mndIsDnodeOnline(pDnode, nowMs)) {
(void)mndSendArbHeartBeatReq(pDnode, arbToken, mndTerm, hbMembers);
TAOS_CHECK_RETURN(mndSendArbHeartBeatReq(pDnode, arbToken, mndTerm, hbMembers));
}
mndReleaseDnode(pMnode, pDnode);
@ -684,7 +686,7 @@ static int32_t mndProcessArbCheckSyncTimer(SRpcMsg *pReq) {
sdbRelease(pSdb, pArbGroup);
}
(void)mndPullupArbUpdateGroupBatch(pMnode, pUpdateArray);
TAOS_CHECK_RETURN(mndPullupArbUpdateGroupBatch(pMnode, pUpdateArray));
taosArrayDestroy(pUpdateArray);
return 0;
@ -795,7 +797,9 @@ _OVER:
if (ret != 0) {
for (size_t i = 0; i < sz; i++) {
SArbGroup *pNewGroup = taosArrayGet(newGroupArray, i);
(void)taosHashRemove(arbUpdateHash, &pNewGroup->vgId, sizeof(pNewGroup->vgId));
if (taosHashRemove(arbUpdateHash, &pNewGroup->vgId, sizeof(pNewGroup->vgId)) != 0) {
mError("failed to remove vgId:%d from arbUpdateHash", pNewGroup->vgId);
}
}
}
@ -839,7 +843,9 @@ static int32_t mndProcessArbUpdateGroupBatchReq(SRpcMsg *pReq) {
SArbGroup *pOldGroup = sdbAcquire(pMnode->pSdb, SDB_ARBGROUP, &newGroup.vgId);
if (!pOldGroup) {
mInfo("vgId:%d, arb skip to update arbgroup, since no obj found", newGroup.vgId);
(void)taosHashRemove(arbUpdateHash, &newGroup.vgId, sizeof(int32_t));
if (taosHashRemove(arbUpdateHash, &newGroup.vgId, sizeof(int32_t)) != 0) {
mError("failed to remove vgId:%d from arbUpdateHash", newGroup.vgId);
}
continue;
}
@ -869,7 +875,9 @@ _OVER:
// failed to update arbgroup
for (size_t i = 0; i < sz; i++) {
SMArbUpdateGroup *pUpdateGroup = taosArrayGet(req.updateArray, i);
(void)taosHashRemove(arbUpdateHash, &pUpdateGroup->vgId, sizeof(int32_t));
if (taosHashRemove(arbUpdateHash, &pUpdateGroup->vgId, sizeof(int32_t)) != 0) {
mError("failed to remove vgId:%d from arbUpdateHash", pUpdateGroup->vgId);
}
}
}
@ -1010,7 +1018,7 @@ static int32_t mndUpdateArbHeartBeat(SMnode *pMnode, int32_t dnodeId, SArray *me
sdbRelease(pMnode->pSdb, pGroup);
}
(void)mndPullupArbUpdateGroupBatch(pMnode, pUpdateArray);
TAOS_CHECK_RETURN(mndPullupArbUpdateGroupBatch(pMnode, pUpdateArray));
taosArrayDestroy(pUpdateArray);
return 0;
@ -1102,7 +1110,7 @@ static int32_t mndProcessArbHbRsp(SRpcMsg *pRsp) {
goto _OVER;
}
(void)mndUpdateArbHeartBeat(pMnode, arbHbRsp.dnodeId, arbHbRsp.hbMembers);
TAOS_CHECK_GOTO(mndUpdateArbHeartBeat(pMnode, arbHbRsp.dnodeId, arbHbRsp.hbMembers), NULL, _OVER);
code = 0;
_OVER:
@ -1249,6 +1257,8 @@ static int32_t mndRetrieveArbGroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
int32_t numOfRows = 0;
int32_t cols = 0;
SArbGroup *pGroup = NULL;
int32_t code = 0;
int32_t lino = 0;
while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_ARBGROUP, pShow->pIter, (void **)&pGroup);
@ -1264,33 +1274,40 @@ static int32_t mndRetrieveArbGroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
sdbRelease(pSdb, pGroup);
continue;
}
char dbNameInGroup[TSDB_DB_FNAME_LEN];
strncpy(dbNameInGroup, pVgObj->dbName, TSDB_DB_FNAME_LEN);
sdbRelease(pSdb, pVgObj);
char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pVgObj->dbName), TSDB_ARB_TOKEN_SIZE + VARSTR_HEADER_SIZE);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false);
STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(dbNameInGroup), TSDB_ARB_TOKEN_SIZE + VARSTR_HEADER_SIZE);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pGroup, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->vgId, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->vgId, false), pGroup, &lino, _OVER);
for (int i = 0; i < TSDB_ARB_GROUP_MEMBER_NUM; i++) {
SArbGroupMember *pMember = &pGroup->members[i];
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pMember->info.dnodeId, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pMember->info.dnodeId, false), pGroup,
&lino, _OVER);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->isSync, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->isSync, false), pGroup, &lino, _OVER);
if (pGroup->assignedLeader.dnodeId != 0) {
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->assignedLeader.dnodeId, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->assignedLeader.dnodeId, false),
pGroup, &lino, _OVER);
char token[TSDB_ARB_TOKEN_SIZE + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(token, pGroup->assignedLeader.token, TSDB_ARB_TOKEN_SIZE + VARSTR_HEADER_SIZE);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)token, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)token, false), pGroup, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->assignedLeader.acked, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pGroup->assignedLeader.acked, false),
pGroup, &lino, _OVER);
} else {
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetNULL(pColInfo, numOfRows);
@ -1305,10 +1322,11 @@ static int32_t mndRetrieveArbGroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
(void)taosThreadMutexUnlock(&pGroup->mutex);
numOfRows++;
sdbRelease(pSdb, pVgObj);
sdbRelease(pSdb, pGroup);
}
_OVER:
if (code != 0) mError("failed to restrieve arb group at line:%d, since %s", lino, tstrerror(code));
pShow->numOfRows += numOfRows;
return numOfRows;

16
source/dnode/mnode/impl/src/mndCompact.c
View File

@ -287,6 +287,8 @@ int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock,
SCompactObj *pCompact = NULL;
char *sep = NULL;
SDbObj *pDb = NULL;
int32_t code = 0;
int32_t lino = 0;
if (strlen(pShow->db) > 0) {
sep = strchr(pShow->db, '.');
@ -310,26 +312,30 @@ int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock,
char tmpBuf[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->compactId, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->compactId, false), pCompact, &lino,
_OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (pDb != NULL || !IS_SYS_DBNAME(pCompact->dbname)) {
SName name = {0};
(void)tNameFromString(&name, pCompact->dbname, T_NAME_ACCT | T_NAME_DB);
TAOS_CHECK_GOTO(tNameFromString(&name, pCompact->dbname, T_NAME_ACCT | T_NAME_DB), &lino, _OVER);
(void)tNameGetDbName(&name, varDataVal(tmpBuf));
} else {
(void)strncpy(varDataVal(tmpBuf), pCompact->dbname, TSDB_SHOW_SQL_LEN);
}
varDataSetLen(tmpBuf, strlen(varDataVal(tmpBuf)));
(void)colDataSetVal(pColInfo, numOfRows, (const char *)tmpBuf, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)tmpBuf, false), pCompact, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->startTime, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->startTime, false), pCompact, &lino,
_OVER);
numOfRows++;
sdbRelease(pSdb, pCompact);
}
_OVER:
if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
pShow->numOfRows += numOfRows;
mndReleaseDb(pMnode, pDb);
return numOfRows;
@ -468,7 +474,7 @@ static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompa
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
*/
}

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

@ -463,6 +463,15 @@ static void freeItem(void *param) {
}
}
#define ADD_TOPIC_TO_ARRAY(element, array) \
char *newTopicCopy = taosStrdup(element); \
MND_TMQ_NULL_CHECK(newTopicCopy);\
if (taosArrayPush(pConsumerNew->array, &newTopicCopy) == NULL){\
taosMemoryFree(newTopicCopy);\
code = terrno;\
goto END;\
}
static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){
int32_t code = 0;
pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *));
@ -477,15 +486,13 @@ static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerOb
if (i >= oldTopicNum) {
void* tmp = taosArrayGetP(pConsumerNew->assignedTopics, j);
MND_TMQ_NULL_CHECK(tmp);
char *newTopicCopy = taosStrdup(tmp);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy));
ADD_TOPIC_TO_ARRAY(tmp, rebNewTopics);
j++;
continue;
} else if (j >= newTopicNum) {
void* tmp = taosArrayGetP(pExistedConsumer->currentTopics, i);
MND_TMQ_NULL_CHECK(tmp);
char *oldTopicCopy = taosStrdup(tmp);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy));
ADD_TOPIC_TO_ARRAY(tmp, rebRemovedTopics);
i++;
continue;
} else {
@ -499,13 +506,11 @@ static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerOb
j++;
continue;
} else if (comp < 0) {
char *oldTopicCopy = taosStrdup(oldTopic);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy));
ADD_TOPIC_TO_ARRAY(oldTopic, rebRemovedTopics);
i++;
continue;
} else {
char *newTopicCopy = taosStrdup(newTopic);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy));
ADD_TOPIC_TO_ARRAY(newTopic, rebNewTopics);
j++;
continue;
}
@ -789,6 +794,9 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
return TSDB_CODE_TMQ_INVALID_MSG;
}
char *pNewTopic = taosStrdup(tmp);
if (pNewTopic == NULL) {
return terrno;
}
removeFromTopicList(pOldConsumer->rebNewTopics, pNewTopic, pOldConsumer->consumerId, "new");
bool existing = existInCurrentTopicList(pOldConsumer, pNewTopic);
if (existing) {

148
source/dnode/mnode/impl/src/mndDb.c
View File

@ -981,7 +981,8 @@ static int32_t mndProcessCreateDbReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
SName name = {0};
(void)tNameFromString(&name, createReq.db, T_NAME_ACCT | T_NAME_DB);
if (tNameFromString(&name, createReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
mError("db:%s, failed to parse db name", createReq.db);
auditRecord(pReq, pMnode->clusterId, "createDB", name.dbname, "", createReq.sql, createReq.sqlLen);
@ -1135,7 +1136,10 @@ static int32_t mndSetAlterDbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
if ((code = sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY)) != 0) {
sdbFreeRaw(pRedoRaw);
TAOS_RETURN(code);
}
return 0;
}
@ -1152,7 +1156,10 @@ static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *p
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
if ((code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY)) != 0) {
sdbFreeRaw(pCommitRaw);
TAOS_RETURN(code);
}
TAOS_RETURN(code);
}
@ -1284,7 +1291,8 @@ static int32_t mndProcessAlterDbReq(SRpcMsg *pReq) {
}
SName name = {0};
(void)tNameFromString(&name, alterReq.db, T_NAME_ACCT | T_NAME_DB);
if (tNameFromString(&name, alterReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
mError("db:%s, failed to parse db name", alterReq.db);
auditRecord(pReq, pMnode->clusterId, "alterDB", name.dbname, "", alterReq.sql, alterReq.sqlLen);
@ -1371,7 +1379,11 @@ static int32_t mndProcessGetDbCfgReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSDbCfgRsp(pRsp, contLen, &cfgRsp);
int32_t ret = 0;
if ((ret = tSerializeSDbCfgRsp(pRsp, contLen, &cfgRsp)) < 0) {
code = ret;
goto _OVER;
}
pReq->info.rsp = pRsp;
pReq->info.rspLen = contLen;
@ -1473,7 +1485,11 @@ static int32_t mndSetDropDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD
sdbRelease(pSdb, pVgroup);
TAOS_RETURN(code);
}
(void)sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPED);
if ((code = sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPED)) != 0) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup);
TAOS_RETURN(code);
}
}
sdbRelease(pSdb, pVgroup);
@ -1498,7 +1514,11 @@ static int32_t mndSetDropDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD
sdbRelease(pSdb, pStbRaw);
return -1;
}
(void)sdbSetRawStatus(pStbRaw, SDB_STATUS_DROPPED);
if ((code = sdbSetRawStatus(pStbRaw, SDB_STATUS_DROPPED)) != 0) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pStbRaw);
return -1;
}
}
sdbRelease(pSdb, pStb);
@ -1562,7 +1582,8 @@ static int32_t mndBuildDropDbRsp(SDbObj *pDb, int32_t *pRspLen, void **ppRsp, bo
TAOS_RETURN(code);
}
(void)tSerializeSDropDbRsp(pRsp, rspLen, &dropRsp);
int32_t ret = 0;
if ((ret = tSerializeSDropDbRsp(pRsp, rspLen, &dropRsp)) < 0) return ret;
*pRspLen = rspLen;
*ppRsp = pRsp;
TAOS_RETURN(code);
@ -1643,7 +1664,8 @@ static int32_t mndProcessDropDbReq(SRpcMsg *pReq) {
}
SName name = {0};
(void)tNameFromString(&name, dropReq.db, T_NAME_ACCT | T_NAME_DB);
if (tNameFromString(&name, dropReq.db, T_NAME_ACCT | T_NAME_DB) < 0)
mError("db:%s, failed to parse db name", dropReq.db);
auditRecord(pReq, pMnode->clusterId, "dropDB", name.dbname, "", dropReq.sql, dropReq.sqlLen);
@ -1805,14 +1827,18 @@ static int32_t mndProcessUseDbReq(SRpcMsg *pReq) {
goto _OVER;
}
(void)tSerializeSUseDbRsp(pRsp, contLen, &usedbRsp);
int32_t ret = 0;
if ((ret = tSerializeSUseDbRsp(pRsp, contLen, &usedbRsp)) < 0) {
code = ret;
goto _OVER;
}
pReq->info.rsp = pRsp;
pReq->info.rspLen = contLen;
_OVER:
if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
mError("db:%s, failed to process use db req since %s", usedbReq.db, terrstr());
mError("db:%s, failed to process use db req since %s", usedbReq.db, tstrerror(code));
}
mndReleaseDb(pMnode, pDb);
@ -1964,7 +1990,8 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbCacheInfo *pDbs, int32_t numOfDbs,
tFreeSDbHbBatchRsp(&batchRsp);
return -1;
}
(void)tSerializeSDbHbBatchRsp(pRsp, rspLen, &batchRsp);
int32_t ret = 0;
if ((ret = tSerializeSDbHbBatchRsp(pRsp, rspLen, &batchRsp)) < 0) return ret;
*ppRsp = pRsp;
*pRspLen = rspLen;
@ -1977,6 +2004,7 @@ static int32_t mndTrimDb(SMnode *pMnode, SDbObj *pDb) {
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
void *pIter = NULL;
int32_t code = 0;
SVTrimDbReq trimReq = {.timestamp = taosGetTimestampSec()};
int32_t reqLen = tSerializeSVTrimDbReq(NULL, 0, &trimReq);
int32_t contLen = reqLen + sizeof(SMsgHead);
@ -1993,7 +2021,11 @@ static int32_t mndTrimDb(SMnode *pMnode, SDbObj *pDb) {
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), contLen, &trimReq);
int32_t ret = 0;
if ((ret = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), contLen, &trimReq)) < 0) {
sdbRelease(pSdb, pVgroup);
return ret;
}
SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
@ -2046,6 +2078,7 @@ static int32_t mndS3MigrateDb(SMnode *pMnode, SDbObj *pDb) {
SVS3MigrateDbReq s3migrateReq = {.timestamp = taosGetTimestampSec()};
int32_t reqLen = tSerializeSVS3MigrateDbReq(NULL, 0, &s3migrateReq);
int32_t contLen = reqLen + sizeof(SMsgHead);
int32_t code = 0;
while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
@ -2061,7 +2094,11 @@ static int32_t mndS3MigrateDb(SMnode *pMnode, SDbObj *pDb) {
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
(void)tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), contLen, &s3migrateReq);
int32_t ret = 0;
if ((ret = tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), contLen, &s3migrateReq)) < 0) {
sdbRelease(pSdb, pVgroup);
return ret;
}
SRpcMsg rpcMsg = {.msgType = TDMT_VND_S3MIGRATE, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
@ -2243,6 +2280,8 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
int32_t cols = 0;
int32_t bytes = pShow->pMeta->pSchemas[cols].bytes;
char *buf = taosMemoryMalloc(bytes);
int32_t code = 0;
int32_t lino = 0;
const char *name = mndGetDbStr(pDb->name);
if (name != NULL) {
@ -2286,47 +2325,47 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
for (int32_t i = 0; i < pShow->numOfColumns; ++i) {
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, i);
if (i == 0) {
(void)colDataSetVal(pColInfo, rows, buf, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, buf, false), &lino, _OVER);
} else if (i == 1) {
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false), &lino, _OVER);
} else if (i == 3) {
(void)colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false), &lino, _OVER);
} else if (i == 14) {
(void)colDataSetVal(pColInfo, rows, precVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, precVstr, false), &lino, _OVER);
} else if (i == 15) {
(void)colDataSetVal(pColInfo, rows, statusVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, statusVstr, false), &lino, _OVER);
} else {
colDataSetNULL(pColInfo, rows);
}
}
} else {
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, buf, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, buf, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->createdTime, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfVgroups, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfVgroups, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&numOfTables, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.replications, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.replications, false), &lino, _OVER);
const char *strictStr = pDb->cfg.strict ? "on" : "off";
char strictVstr[24] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(strictVstr, strictStr, 24);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)strictVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)strictVstr, false), &lino, _OVER);
char durationVstr[128] = {0};
int32_t len = formatDurationOrKeep(&durationVstr[VARSTR_HEADER_SIZE], pDb->cfg.daysPerFile);
varDataSetLen(durationVstr, len);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)durationVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)durationVstr, false), &lino, _OVER);
char keepVstr[512] = {0};
char keep0Str[128] = {0};
@ -2344,67 +2383,67 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
}
varDataSetLen(keepVstr, len);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)keepVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)keepVstr, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.buffer, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.buffer, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pageSize, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pageSize, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pages, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.pages, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.minRows, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.minRows, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.maxRows, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.maxRows, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.compression, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.compression, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)precVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)precVstr, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)statusVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)statusVstr, false), &lino, _OVER);
char *rentensionVstr = buildRetension(pDb->cfg.pRetensions);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (rentensionVstr == NULL) {
colDataSetNULL(pColInfo, rows);
} else {
(void)colDataSetVal(pColInfo, rows, (const char *)rentensionVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)rentensionVstr, false), &lino, _OVER);
taosMemoryFree(rentensionVstr);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false), &lino, _OVER);
const char *cacheModelStr = getCacheModelStr(pDb->cfg.cacheLast);
char cacheModelVstr[24] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(cacheModelVstr, cacheModelStr, 24);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)cacheModelVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)cacheModelVstr, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.cacheLastSize, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.cacheLastSize, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walLevel, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walLevel, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walFsyncPeriod, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walFsyncPeriod, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionPeriod, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionPeriod, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionSize, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.walRetentionSize, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.sstTrigger, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.sstTrigger, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
int16_t hashPrefix = pDb->cfg.hashPrefix;
@ -2413,39 +2452,40 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
} else if (hashPrefix < 0) {
hashPrefix = pDb->cfg.hashPrefix + strlen(pDb->name) + 1;
}
(void)colDataSetVal(pColInfo, rows, (const char *)&hashPrefix, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&hashPrefix, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.hashSuffix, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.hashSuffix, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.tsdbPageSize, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.tsdbPageSize, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.keepTimeOffset, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.keepTimeOffset, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.s3ChunkSize, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.s3ChunkSize, false), &lino, _OVER);
char keeplocalVstr[128] = {0};
len = sprintf(&keeplocalVstr[VARSTR_HEADER_SIZE], "%dm", pDb->cfg.s3KeepLocal);
varDataSetLen(keeplocalVstr, len);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)keeplocalVstr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)keeplocalVstr, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.s3Compact, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.s3Compact, false), &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.withArbitrator, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)&pDb->cfg.withArbitrator, false), &lino, _OVER);
const char *encryptAlgorithmStr = getEncryptAlgorithmStr(pDb->cfg.encryptAlgorithm);
char encryptAlgorithmVStr[24] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(encryptAlgorithmVStr, encryptAlgorithmStr, 24);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, rows, (const char *)encryptAlgorithmVStr, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, rows, (const char *)encryptAlgorithmVStr, false), &lino, _OVER);
}
_OVER:
if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
taosMemoryFree(buf);
}

34
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -1694,6 +1694,8 @@ static int32_t mndRetrieveConfigs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
char cfgVals[TSDB_CONFIG_NUMBER][TSDB_CONFIG_VALUE_LEN + 1] = {0};
char *pWrite = NULL;
int32_t cols = 0;
int32_t code = 0;
int32_t lino = 0;
cfgOpts[totalRows] = "statusInterval";
(void)snprintf(cfgVals[totalRows], TSDB_CONFIG_VALUE_LEN, "%d", tsStatusInterval);
@ -1741,15 +1743,17 @@ static int32_t mndRetrieveConfigs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
STR_WITH_MAXSIZE_TO_VARSTR(buf, cfgOpts[i], TSDB_CONFIG_OPTION_LEN);
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)buf, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)buf, false), &lino, _OVER);
STR_WITH_MAXSIZE_TO_VARSTR(bufVal, cfgVals[i], TSDB_CONFIG_VALUE_LEN);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)bufVal, false);
TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)bufVal, false), &lino, _OVER);
numOfRows++;
}
_OVER:
if (code != 0) mError("failed to retrieve configs at line:%d since %s", lino, tstrerror(code));
pShow->numOfRows += numOfRows;
return numOfRows;
}
@ -1765,6 +1769,8 @@ static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
SDnodeObj *pDnode = NULL;
int64_t curMs = taosGetTimestampMs();
char buf[TSDB_EP_LEN + VARSTR_HEADER_SIZE];
int32_t code = 0;
int32_t lino = 0;
while (numOfRows < rows) {
pShow->pIter = sdbFetchAll(pSdb, SDB_DNODE, pShow->pIter, (void **)&pDnode, &objStatus, true);
@ -1774,19 +1780,20 @@ static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
cols = 0;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->id, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->id, false), pDnode, &lino, _OVER);
STR_WITH_MAXSIZE_TO_VARSTR(buf, pDnode->ep, pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, buf, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, buf, false), pDnode, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
int16_t id = mndGetVnodesNum(pMnode, pDnode->id);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&id, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&id, false), pDnode, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->numOfSupportVnodes, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->numOfSupportVnodes, false), pDnode,
&lino, _OVER);
const char *status = "ready";
if (objStatus == SDB_STATUS_CREATING) status = "creating";
@ -1802,31 +1809,36 @@ static int32_t mndRetrieveDnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
STR_TO_VARSTR(buf, status);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, buf, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, buf, false), pDnode, &lino, _OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->createdTime, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->createdTime, false), pDnode, &lino,
_OVER);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->rebootTime, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pDnode->rebootTime, false), pDnode, &lino,
_OVER);
char *b = taosMemoryCalloc(VARSTR_HEADER_SIZE + strlen(offlineReason[pDnode->offlineReason]) + 1, 1);
STR_TO_VARSTR(b, online ? "" : offlineReason[pDnode->offlineReason]);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, b, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, b, false), pDnode, &lino, _OVER);
taosMemoryFreeClear(b);
#ifdef TD_ENTERPRISE
STR_TO_VARSTR(buf, pDnode->machineId);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
(void)colDataSetVal(pColInfo, numOfRows, buf, false);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, buf, false), pDnode, &lino, _OVER);
#endif
numOfRows++;
sdbRelease(pSdb, pDnode);
}
_OVER:
if (code != 0) mError("failed to retrieve dnodes at line:%d since %s", lino, tstrerror(code));
pShow->numOfRows += numOfRows;
return numOfRows;
}

Some files were not shown because too many files have changed in this diff Show More