Skip to content

T
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 322b8c30 编写于 作者: G gccgdb1234
浏览文件
操作

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

上级 253257e4 7d015bd7
展开全部 隐藏空白更改
内联 并排
Showing with 718 addition and 4546 deletion
Jenkinsfile2
浏览文件 @ 322b8c30
......@@ -309,6 +309,7 @@ def pre_test_build_win() {
'''
bat '''
cd %WIN_CONNECTOR_ROOT%
python.exe -m pip install --upgrade pip
python -m pip install .
xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32
'''
......@@ -327,6 +328,7 @@ def run_win_test() {
bat '''
echo "windows test ..."
cd %WIN_CONNECTOR_ROOT%
python.exe -m pip install --upgrade pip
python -m pip install .
xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32
ls -l C:\\Windows\\System32\\taos.dll
......
cmake/taosadapter_CMakeLists.txt.in
浏览文件 @ 322b8c30
......@@ -2,7 +2,7 @@
# taosadapter
ExternalProject_Add(taosadapter
GIT_REPOSITORY https://github.com/taosdata/taosadapter.git
GIT_TAG ed6a160
GIT_TAG 3d21433
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosadapter"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
cmake/taostools_CMakeLists.txt.in
浏览文件 @ 322b8c30
......@@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG 2d68404
GIT_TAG 57bdfbf
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
cmake/taosws_CMakeLists.txt.in
浏览文件 @ 322b8c30
......@@ -2,7 +2,7 @@
# taosws-rs
ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG 97c4bac
GIT_TAG 7a54d21
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
docs/en/14-reference/08-taos-shell.md
浏览文件 @ 322b8c30
......@@ -47,27 +47,28 @@ If the displayed content is followed by `...` you can use this command to change
You can change the behavior of TDengine CLI by specifying command-line parameters. The following parameters are commonly used.
- -h, --host=HOST: FQDN of the server where the TDengine server is to be connected. Default is to connect to the local service
- -P, --port=PORT: Specify the port number to be used by the server. Default is `6030`
- -u, --user=USER: the user name to use when connecting. Default is `root`
- -p, --password=PASSWORD: the password to use when connecting to the server. Default is `taosdata`
- -h HOST: FQDN of the server where the TDengine server is to be connected. Default is to connect to the local service
- -P PORT: Specify the port number to be used by the server. Default is `6030`
- -u USER: the user name to use when connecting. Default is `root`
- -p PASSWORD: the password to use when connecting to the server. Default is `taosdata`
- -?, --help: print out all command-line arguments
And many more parameters.
- -c, --config-dir: Specify the directory where configuration file exists. The default is `/etc/taos`, and the default name of the configuration file in this directory is `taos.cfg`
- -C, --dump-config: Print the configuration parameters of `taos.cfg` in the default directory or specified by -c
- -d, --database=DATABASE: Specify the database to use when connecting to the server
- -D, --directory=DIRECTORY: Import the SQL script file in the specified path
- -f, --file=FILE: Execute the SQL script file in non-interactive mode
- -k, --check=CHECK: Specify the table to be checked
- -l, --pktlen=PKTLEN: Test package size to be used for network testing
- -n, --netrole=NETROLE: test scope for network connection test, default is `startup`. The value can be `client`, `server`, `rpc`, `startup`, `sync`, `speed`, or `fqdn`.
- -r, --raw-time: output the timestamp format as unsigned 64-bits integer (uint64_t in C language)
- -s, --commands=COMMAND: execute SQL commands in non-interactive mode
- -S, --pkttype=PKTTYPE: Specify the packet type used for network testing. The default is TCP, can be specified as either TCP or UDP when `speed` is specified to `netrole` parameter
- -T, --thread=THREADNUM: The number of threads to import data in multi-threaded mode
- -s, --commands: Run TDengine CLI commands without entering the terminal
- -a AUTHSTR: The auth string to use when connecting to the server
- -A: Generate auth string from password
- -c CONFIGDIR: Specify the directory where configuration file exists. The default is `/etc/taos`, and the default name of the configuration file in this directory is `taos.cfg`
- -C: Print the configuration parameters of `taos.cfg` in the default directory or specified by -c
- -d DATABASE: Specify the database to use when connecting to the server
- -f FILE: Execute the SQL script file in non-interactive mode
- -k: Check the service status, 0: unavailable，1: network ok，2: service ok，3: service degraded，4: exiting
- -l PKTLEN: Test package length to be used for network testing
- -n NETROLE: test scope for network connection test, default is `client`. The value can be `client`, `server`
- -N PKTNUM: Test package numbers to be used for network testing
- -r: output the timestamp format as unsigned 64-bits integer (uint64_t in C language)
- -s COMMAND: execute SQL commands in non-interactive mode
- -t: Check the details of the service status，status same as -k
- -w DISPLAYWIDTH: 客户端列显示宽度
- -z, --timezone=TIMEZONE: Specify time zone. Default is the value of current configuration file
- -V, --version: Print out the current version number
......
docs/en/14-reference/09-support-platform/index.md
浏览文件 @ 322b8c30
......@@ -5,12 +5,11 @@ description: "List of platforms supported by TDengine server, client, and connec
## List of supported platforms for TDengine server
| | **CentOS 7/8** | **Ubuntu 16/18/20** | **Other Linux** |
| ------------ | -------------- | ------------------- | --------------- |
| X64 | ● | ● | |
| MIPS64 | | | ● |
| ARM64 | | ○ | ○ |
| Alpha64 | | | ○ |
| | **Windows 10/11** | **CentOS 7.9/8** | **Ubuntu 18/20** | **Other Linux** | **UOS** | **Kylin** | **Ningsi V60/V80** | **HUAWEI EulerOS** |
| ------------------ | ----------------- | ---------------- | ---------------- | --------------- | ------- | --------- | ------------------ | ------------------ |
| X64 | ● | ● | ● | | ● | ● | ● | |
| Raspberry Pi ARM64 | | | | ● | | | | |
| HUAWEI cloud ARM64 | | | | | | | | ● |
Note: ● means officially tested and verified, ○ means unofficially tested and verified.
......
docs/en/14-reference/12-directory.md
浏览文件 @ 322b8c30
......@@ -25,7 +25,6 @@ All executable files of TDengine are in the _/usr/local/taos/bin_ directory by d
- _taosBenchmark_: TDengine testing tool
- _remove.sh_: script to uninstall TDengine, please execute it carefully, link to the **rmtaos** command in the /usr/bin directory. Will remove the TDengine installation directory `/usr/local/taos`, but will keep `/etc/taos`, `/var/lib/taos`, `/var/log/taos`
- _taosadapter_: server-side executable that provides RESTful services and accepts writing requests from a variety of other softwares
- _tarbitrator_: provides arbitration for two-node cluster deployments
- _TDinsight.sh_: script to download TDinsight and install it
- _set_core.sh_: script for setting up the system to generate core dump files for easy debugging
- _taosd-dump-cfg.gdb_: script to facilitate debugging of taosd's gdb execution.
......
docs/zh/05-get-started/01-docker.md
浏览文件 @ 322b8c30
......@@ -2,6 +2,9 @@
sidebar_label: Docker
title: 通过 Docker 快速体验 TDengine
---
:::info
如果您希望对 TDengine 贡献代码或对内部实现感兴趣，请参考我们的 [TDengine GitHub 主页](https://github.com/taosdata/TDengine) 下载源码构建和安装.
:::
本节首先介绍如何通过 Docker 快速体验 TDengine，然后介绍如何在 Docker 环境下体验 TDengine 的写入和查询功能。
......
docs/zh/14-reference/03-connector/03-connector.mdx
浏览文件 @ 322b8c30
......@@ -10,14 +10,14 @@ TDengine 提供了丰富的应用程序开发接口，为了便于用户快速
目前 TDengine 的原生接口连接器可支持的平台包括：X64/X86/ARM64/ARM32/MIPS/Alpha 等硬件平台，以及 Linux/Win64/Win32 等开发环境。对照矩阵如下：
| **CPU** | **OS** | **JDBC** | **Python** | **Go** | **Node.js** | **C#** | **Rust** | C/C++ |
| **CPU** | **OS** | **Java** | **Python** | **Go** | **Node.js** | **C#** | **Rust** | C/C++ |
| -------------- | --------- | -------- | ---------- | ------ | ----------- | ------ | -------- | ----- |
| **X86 64bit** | **Linux** | ● | ● | ● | ● | ● | ● | ● |
| **X86 64bit** | **Win64** | ● | ● | ● | ● | ● | ● | ● |
| **X86 64bit** | **Win32** | ● | ● | ● | ● | ○ | ○ | ● |
| **X86 32bit** | **Win32** | ○ | ○ | ○ | ○ | ○ | ○ | ● |
| **ARM64** | **Linux** | ● | ● | ● | ● | ○ | ○ | ● |
| **ARM32** | **Linux** | ● | ● | ● | ● | ○ | ○ | ● |
| **ARM32** | **Linux** | ○ | ○ | ○ | ○ | ○ | ○ | ● |
| **MIPS 龙芯** | **Linux** | ○ | ○ | ○ | ○ | ○ | ○ | ○ |
| **Alpha 申威** | **Linux** | ○ | ○ | -- | -- | -- | -- | ○ |
| **X86 海光** | **Linux** | ○ | ○ | ○ | -- | -- | -- | ○ |
......@@ -32,6 +32,7 @@ TDengine 版本更新往往会增加新的功能特性，列表中的连接器
| **TDengine 版本** | **Java** | **Python** | **Go** | **C#** | **Node.js** | **Rust** |
| --------------------- | -------- | ---------- | ------------ | ------------- | --------------- | -------- |
| **3.0.0.0 及以上** | 3.0.0 | 当前版本 | 3.0 分支 | 3.0.0 | 3.0.0 | 当前版本 |
| **2.4.0.14 及以上** | 2.0.38 | 当前版本 | develop 分支 | 1.0.2 - 1.0.6 | 2.0.10 - 2.0.12 | 当前版本 |
| **2.4.0.6 及以上** | 2.0.37 | 当前版本 | develop 分支 | 1.0.2 - 1.0.6 | 2.0.10 - 2.0.12 | 当前版本 |
| **2.4.0.4 - 2.4.0.5** | 2.0.37 | 当前版本 | develop 分支 | 1.0.2 - 1.0.6 | 2.0.10 - 2.0.12 | 当前版本 |
......@@ -48,9 +49,8 @@ TDengine 版本更新往往会增加新的功能特性，列表中的连接器
| -------------- | -------- | ---------- | ------ | ------ | ----------- | -------- |
| **连接管理** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **普通查询** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **连续查询** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **参数绑定** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **订阅功能** | 支持 | 支持 | 支持 | 支持 | 支持 | 暂不支持 |
| ** TMQ ** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **Schemaless** | 支持 | 支持 | 支持 | 支持 | 支持 | 支持 |
| **DataFrame** | 不支持 | 支持 | 不支持 | 不支持 | 不支持 | 不支持 |
......@@ -58,17 +58,17 @@ TDengine 版本更新往往会增加新的功能特性，列表中的连接器
由于不同编程语言数据库框架规范不同，并不意味着所有 C/C++ 接口都需要对应封装支持。
:::
### 使用 REST 接口
### 使用 http (REST 或 WebSocket) 接口
| **功能特性** | **Java** | **Python** | **Go** | **C#（暂不支持）** | **Node.js** | **Rust** |
| ------------------------------ | -------- | ---------- | -------- | ------------------ | ----------- | -------- |
| **连接管理** | 支持 | 支持 | 支持 | N/A | 支持 | 支持 |
| **普通查询** | 支持 | 支持 | 支持 | N/A | 支持 | 支持 |
| **连续查询** | 支持 | 支持 | 支持 | N/A | 支持 | 支持 |
| **参数绑定** | 不支持 | 不支持 | 不支持 | N/A | 不支持 | 不支持 |
| **订阅功能** | 不支持 | 不支持 | 不支持 | N/A | 不支持 | 不支持 |
| **Schemaless** | 暂不支持 | 暂不支持 | 暂不支持 | N/A | 暂不支持 | 暂不支持 |
| **批量拉取（基于 WebSocket）** | 支持 | 暂不支持 | 暂不支持 | N/A | 暂不支持 | 暂不支持 |
| **参数绑定** | 不支持 | 暂不支持 | 暂不支持 | N/A | 不支持 | 支持 |
| ** TMQ ** | 不支持 | 暂不支持 | 暂不支持 | N/A | 不支持 | 支持 |
| **Schemaless** | 暂不支持 | 暂不支持 | 暂不支持 | N/A | 不支持 | 暂不支持 |
| **批量拉取（基于 WebSocket）** | 支持 | 支持 | 暂不支持 | N/A | 不支持 | 支持 |
| **DataFrame** | 不支持 | 支持 | 不支持 | N/A | 不支持 | 不支持 |
:::warning
......
docs/zh/14-reference/08-taos-shell.md
浏览文件 @ 322b8c30
......@@ -48,29 +48,30 @@ taos> SET MAX_BINARY_DISPLAY_WIDTH <nn>;
您可通过配置命令行参数来改变 TDengine CLI 的行为。以下为常用的几个命令行参数：
- -h, --host=HOST: 要连接的 TDengine 服务端所在服务器的 FQDN, 默认为连接本地服务
- -P, --port=PORT: 指定服务端所用端口号
- -u, --user=USER: 连接时使用的用户名
- -p, --password=PASSWORD: 连接服务端时使用的密码
- -h HOST: 要连接的 TDengine 服务端所在服务器的 FQDN, 默认为连接本地服务
- -P PORT: 指定服务端所用端口号
- -u USER: 连接时使用的用户名
- -p PASSWORD: 连接服务端时使用的密码
- -?, --help: 打印出所有命令行参数
还有更多其他参数：
- -c, --config-dir: 指定配置文件目录，Linux 环境下默认为 `/etc/taos`，该目录下的配置文件默认名称为 `taos.cfg`
- -C, --dump-config: 打印 -c 指定的目录中 `taos.cfg` 的配置参数
- -d, --database=DATABASE: 指定连接到服务端时使用的数据库
- -D, --directory=DIRECTORY: 导入指定路径中的 SQL 脚本文件
- -f, --file=FILE: 以非交互模式执行 SQL 脚本文件。文件中一个 SQL 语句只能占一行
- -k, --check=CHECK: 指定要检查的表
- -l, --pktlen=PKTLEN: 网络测试时使用的测试包大小
- -n, --netrole=NETROLE: 网络连接测试时的测试范围，默认为 `startup`, 可选值为 `client`、`server`、`rpc`、`startup`、`sync`、`speed` 和 `fqdn` 之一
- -r, --raw-time: 将时间输出出无符号 64 位整数类型(即 C 语音中 uint64_t)
- -s, --commands=COMMAND: 以非交互模式执行的 SQL 命令
- -S, --pkttype=PKTTYPE: 指定网络测试所用的包类型，默认为 TCP。只有 netrole 为 `speed` 时既可以指定为 TCP 也可以指定为 UDP
- -T, --thread=THREADNUM: 以多线程模式导入数据时的线程数
- -s, --commands: 在不进入终端的情况下运行 TDengine 命令
- -z, --timezone=TIMEZONE: 指定时区，默认为本地时区
- -V, --version: 打印出当前版本号
- -a AUTHSTR: 连接服务端的授权信息
- -A: 通过用户名和密码计算授权信息
- -c CONFIGDIR: 指定配置文件目录，Linux 环境下默认为 `/etc/taos`，该目录下的配置文件默认名称为 `taos.cfg`
- -C: 打印 -c 指定的目录中 `taos.cfg` 的配置参数
- -d DATABASE: 指定连接到服务端时使用的数据库
- -f FILE: 以非交互模式执行 SQL 脚本文件。文件中一个 SQL 语句只能占一行
- -k: 测试服务端运行状态，0: unavailable，1: network ok，2: service ok，3: service degraded，4: exiting
- -l PKTLEN: 网络测试时使用的测试包大小
- -n NETROLE: 网络连接测试时的测试范围，默认为 `client`, 可选值为 `client`、`server`
- -N PKTNUM: 网络测试时使用的测试包数量
- -r: 将时间输出出无符号 64 位整数类型(即 C 语音中 uint64_t)
- -s COMMAND: 以非交互模式执行的 SQL 命令
- -t: 测试服务端启动状态，状态同-k
- -w DISPLAYWIDTH: 客户端列显示宽度
- -z TIMEZONE: 指定时区，默认为本地时区
- -V: 打印出当前版本号
示例：
......
docs/zh/14-reference/09-support-platform/index.md
浏览文件 @ 322b8c30
......@@ -5,18 +5,11 @@ description: "TDengine 服务端、客户端和连接器支持的平台列表"
## TDengine 服务端支持的平台列表
| | **CentOS 7/8** | **Ubuntu 16/18/20** | **Other Linux** | **统信 UOS** | **银河/中标麒麟** | **凝思 V60/V80** | **华为 EulerOS** |
| ------------ | -------------- | ------------------- | --------------- | ------------ | ----------------- | ---------------- | ---------------- |
| X64 | ● | ● | | ○ | ● | ● | ● |
| 龙芯 MIPS64 | | | ● | | | | |
| 鲲鹏 ARM64 | | ○ | ○ | | ● | | |
| 申威 Alpha64 | | | ○ | ● | | | |
| 飞腾 ARM64 | | ○ 优麒麟 | | | | | |
| 海光 X64 | ● | ● | ● | ○ | ● | ● | |
| 瑞芯微 ARM64 | | | ○ | | | | |
| 全志 ARM64 | | | ○ | | | | |
| 炬力 ARM64 | | | ○ | | | | |
| 华为云 ARM64 | | | | | | | ● |
| | **Windows 10/11** | **CentOS 7.9/8** | **Ubuntu 18/20** | **Other Linux** | **统信 UOS** | **银河/中标麒麟** | **凝思 V60/V80** | **华为 EulerOS** |
| ------------ | ----------------- | ---------------- | ---------------- | --------------- | ------------ | ----------------- | ---------------- | ---------------- |
| X64 | ● | ● | ● | | ● | ● | ● | |
| 树莓派 ARM64 | | | | ● | | | | |
| 华为云 ARM64 | | | | | | | | ● |
注： ● 表示经过官方测试验证， ○ 表示非官方测试验证。
......
docs/zh/14-reference/12-directory.md
浏览文件 @ 322b8c30
......@@ -25,7 +25,6 @@ TDengine 的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下
- _taosBenchmark_：TDengine 测试工具
- _remove.sh_：卸载 TDengine 的脚本，请谨慎执行，链接到/usr/bin 目录下的**rmtaos**命令。会删除 TDengine 的安装目录/usr/local/taos，但会保留/etc/taos、/var/lib/taos、/var/log/taos
- _taosadapter_: 提供 RESTful 服务和接受其他多种软件写入请求的服务端可执行文件
- _tarbitrator_: 提供双节点集群部署的仲裁功能
- _TDinsight.sh_：用于下载 TDinsight 并安装的脚本
- _set_core.sh_：用于方便调试设置系统生成 core dump 文件的脚本
- _taosd-dump-cfg.gdb_：用于方便调试 taosd 的 gdb 执行脚本。
......
rust @ 7ed7a977
Subproject commit 7ed7a97715388fa144718764d6bf20f9bfc29a12
include/client/taos.h
浏览文件 @ 322b8c30
......@@ -196,15 +196,6 @@ DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, vo
DLL_EXPORT void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param);
DLL_EXPORT const void *taos_get_raw_block(TAOS_RES *res);
// Shuduo: temporary enable for app build
#if 1
typedef void (*__taos_sub_fn_t)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char *topic, const char *sql, __taos_sub_fn_t fp,
void *param, int interval);
DLL_EXPORT TAOS_RES *taos_consume(TAOS_SUB *tsub);
DLL_EXPORT void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress);
#endif
DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList);
DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
......@@ -281,10 +272,6 @@ DLL_EXPORT const char *tmq_get_table_name(TAOS_RES *res);
/* ------------------------------ TMQ END -------------------------------- */
#if 1 // Shuduo: temporary enable for app build
typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
#endif
typedef enum {
TSDB_SRV_STATUS_UNAVAILABLE = 0,
TSDB_SRV_STATUS_NETWORK_OK = 1,
......
include/common/tdataformat.h
浏览文件 @ 322b8c30
......@@ -200,8 +200,6 @@ struct STag {
#if 1 //================================================================================================================================================
// Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap.
#define TD_SUPPORT_BITMAP
#define TD_SUPPORT_READ2
#define TD_SUPPORT_BACK2 // suppport back compatibility of 2.0
#define TASSERT(x) ASSERT(x)
......
include/common/tmsg.h
浏览文件 @ 322b8c30
......@@ -1369,8 +1369,8 @@ typedef struct {
int32_t numOfCols;
int64_t skey;
int64_t ekey;
int64_t version; // for stream
TSKEY watermark;// for stream
int64_t version; // for stream
TSKEY watermark; // for stream
char data[];
} SRetrieveTableRsp;
......@@ -3080,6 +3080,22 @@ typedef struct SDeleteRes {
int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes);
int32_t tDecodeDeleteRes(SDecoder* pCoder, SDeleteRes* pRes);
typedef struct {
int64_t uid;
int64_t ts;
} SSingleDeleteReq;
int32_t tEncodeSSingleDeleteReq(SEncoder* pCoder, const SSingleDeleteReq* pReq);
int32_t tDecodeSSingleDeleteReq(SDecoder* pCoder, SSingleDeleteReq* pReq);
typedef struct {
int64_t suid;
SArray* deleteReqs; // SArray<SSingleDeleteReq>
} SBatchDeleteReq;
int32_t tEncodeSBatchDeleteReq(SEncoder* pCoder, const SBatchDeleteReq* pReq);
int32_t tDecodeSBatchDeleteReq(SDecoder* pCoder, SBatchDeleteReq* pReq);
typedef struct {
int32_t msgIdx;
int32_t msgType;
......
include/common/tmsgdef.h
浏览文件 @ 322b8c30
......@@ -202,6 +202,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_SUBMIT_RSMA, "vnode-submit-rsma", SSubmitReq, SSubmitRsp)
TD_DEF_MSG_TYPE(TDMT_VND_FETCH_RSMA, "vnode-fetch-rsma", SRSmaFetchMsg, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "delete-data", SVDeleteReq, SVDeleteRsp)
TD_DEF_MSG_TYPE(TDMT_VND_BATCH_DEL, "batch-delete", SBatchDeleteReq, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "alter-config", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_REPLICA, "alter-replica", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIRM, "alter-confirm", NULL, NULL)
......
include/common/trow.h
浏览文件 @ 322b8c30
......@@ -319,21 +319,17 @@ typedef struct {
col_id_t kvIdx; // [0, nKvCols)
} STSRowIter;
void tdSTSRowIterReset(STSRowIter *pIter, STSRow *pRow);
void tdSTSRowIterInit(STSRowIter *pIter, STSchema *pSchema);
void tdSTSRowIterInit(STSRowIter *pIter, STSchema *pSchema);
void tdSTSRowIterReset(STSRowIter *pIter, STSRow *pRow);
bool tdSTSRowIterFetch(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal);
bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal);
int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow);
bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal);
bool tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colType, int32_t offset, SCellVal *pVal);
bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType, col_id_t *nIdx, SCellVal *pVal);
bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal);
bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t flen, uint32_t offset, col_id_t colIdx,
SCellVal *pVal);
bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVal);
void tdSCellValPrint(SCellVal *pVal, int8_t colType);
void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag);
#ifdef __cplusplus
}
#endif
#endif /*_TD_COMMON_ROW_H_*/
#endif /*_TD_COMMON_ROW_H_*/
\ No newline at end of file
include/libs/function/function.h
浏览文件 @ 322b8c30
......@@ -54,10 +54,6 @@ typedef struct SFuncExecFuncs {
FExecCombine combine;
} SFuncExecFuncs;
typedef struct SFileBlockInfo {
int32_t numBlocksOfStep;
} SFileBlockInfo;
#define MAX_INTERVAL_TIME_WINDOW 1000000 // maximum allowed time windows in final results
#define TOP_BOTTOM_QUERY_LIMIT 100
......@@ -171,8 +167,6 @@ typedef struct tExprNode {
};
} tExprNode;
void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *));
struct SScalarParam {
bool colAlloced;
SColumnInfoData *columnData;
......@@ -182,14 +176,10 @@ struct SScalarParam {
int32_t numOfRows;
};
int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, SResultDataInfo* pInfo, int16_t extLength,
bool isSuperTable);
void resetResultRowEntryResult(SqlFunctionCtx* pCtx, int32_t num);
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock);
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
typedef struct SPoint {
int64_t key;
......
include/util/taoserror.h
浏览文件 @ 322b8c30
......@@ -284,11 +284,13 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_CONSUMER_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x03EA)
#define TSDB_CODE_MND_TOPIC_SUBSCRIBED TAOS_DEF_ERROR_CODE(0, 0x03EB)
#define TSDB_CODE_MND_CGROUP_USED TAOS_DEF_ERROR_CODE(0, 0x03EC)
#define TSDB_CODE_MND_TOPIC_MUST_BE_DELETED TAOS_DEF_ERROR_CODE(0, 0x03ED)
// mnode-stream
#define TSDB_CODE_MND_STREAM_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03F0)
#define TSDB_CODE_MND_STREAM_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03F1)
#define TSDB_CODE_MND_INVALID_STREAM_OPTION TAOS_DEF_ERROR_CODE(0, 0x03F2)
#define TSDB_CODE_MND_STREAM_MUST_BE_DELETED TAOS_DEF_ERROR_CODE(0, 0x03F3)
// mnode-sma
#define TSDB_CODE_MND_SMA_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0480)
......
source/client/src/TSDBJNIConnector.c
浏览文件 @ 322b8c30
......@@ -611,65 +611,6 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_closeConnectionIm
}
}
JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp(JNIEnv *env, jobject jobj, jlong con,
jboolean restart, jstring jtopic,
jstring jsql, jint jinterval) {
jlong sub = 0;
TAOS *taos = (TAOS *)con;
char *topic = NULL;
char *sql = NULL;
jniGetGlobalMethod(env);
jniDebug("jobj:%p, in TSDBJNIConnector_subscribeImp", jobj);
if (jtopic != NULL) {
topic = (char *)(*env)->GetStringUTFChars(env, jtopic, NULL);
}
if (jsql != NULL) {
sql = (char *)(*env)->GetStringUTFChars(env, jsql, NULL);
}
if (topic == NULL || sql == NULL) {
jniDebug("jobj:%p, invalid argument: topic or sql is NULL", jobj);
return sub;
}
TAOS_SUB *tsub = taos_subscribe(taos, (int)restart, topic, sql, NULL, NULL, jinterval);
sub = (jlong)tsub;
if (sub == 0) {
jniDebug("jobj:%p, failed to subscribe: topic:%s", jobj, topic);
} else {
jniDebug("jobj:%p, successfully subscribe: topic: %s", jobj, topic);
}
(*env)->ReleaseStringUTFChars(env, jtopic, topic);
(*env)->ReleaseStringUTFChars(env, jsql, sql);
return sub;
}
JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNIEnv *env, jobject jobj, jlong sub) {
jniDebug("jobj:%p, in TSDBJNIConnector_consumeImp, sub:%lld", jobj, sub);
jniGetGlobalMethod(env);
TAOS_SUB *tsub = (TAOS_SUB *)sub;
TAOS_RES *res = taos_consume(tsub);
if (res == NULL) {
jniError("jobj:%p, tsub:%p, taos_consume returns NULL", jobj, tsub);
return 0l;
}
return (jlong)res;
}
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_unsubscribeImp(JNIEnv *env, jobject jobj, jlong sub,
jboolean keepProgress) {
TAOS_SUB *tsub = (TAOS_SUB *)sub;
taos_unsubscribe(tsub, keepProgress);
}
JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_validateCreateTableSqlImp(JNIEnv *env, jobject jobj,
jlong con, jbyteArray jsql) {
TAOS *tscon = (TAOS *)con;
......@@ -1087,4 +1028,4 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_schemalessInsert
return JNI_OUT_OF_MEMORY;
}
return (jlong)tres;
}
\ No newline at end of file
}
source/client/src/clientMain.c
浏览文件 @ 322b8c30
......@@ -939,21 +939,6 @@ const void *taos_get_raw_block(TAOS_RES *res) {
return pRequest->body.resInfo.pData;
}
TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char *topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp,
void *param, int interval) {
// TODO
return NULL;
}
TAOS_RES *taos_consume(TAOS_SUB *tsub) {
// TODO
return NULL;
}
void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
// TODO
}
int taos_load_table_info(TAOS *taos, const char *tableNameList) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
......
source/client/src/tmq.c
浏览文件 @ 322b8c30
......@@ -204,7 +204,7 @@ typedef struct {
typedef struct {
SMqCommitCbParamSet* params;
STqOffset* pOffset;
} SMqCommitCbParam2;
} SMqCommitCbParam;
tmq_conf_t* tmq_conf_new() {
tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
......@@ -212,6 +212,7 @@ tmq_conf_t* tmq_conf_new() {
conf->autoCommit = true;
conf->autoCommitInterval = 5000;
conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
conf->hbBgEnable = true;
return conf;
}
......@@ -368,8 +369,8 @@ static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
return sprintf(dst, "%s:%d", topicName, vg);
}
int32_t tmqCommitCb2(void* param, SDataBuf* pBuf, int32_t code) {
SMqCommitCbParam2* pParam = (SMqCommitCbParam2*)param;
int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
SMqCommitCbParamSet* pParamSet = (SMqCommitCbParamSet*)pParam->params;
// push into array
#if 0
......@@ -440,7 +441,7 @@ static int32_t tmqSendCommitReq(tmq_t* tmq, SMqClientVg* pVg, SMqClientTopic* pT
tEncodeSTqOffset(&encoder, pOffset);
// build param
SMqCommitCbParam2* pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam2));
SMqCommitCbParam* pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam));
pParam->params = pParamSet;
pParam->pOffset = pOffset;
......@@ -465,7 +466,7 @@ static int32_t tmqSendCommitReq(tmq_t* tmq, SMqClientVg* pVg, SMqClientTopic* pT
pMsgSendInfo->requestObjRefId = 0;
pMsgSendInfo->param = pParam;
pMsgSendInfo->paramFreeFp = taosMemoryFree;
pMsgSendInfo->fp = tmqCommitCb2;
pMsgSendInfo->fp = tmqCommitCb;
pMsgSendInfo->msgType = TDMT_VND_MQ_COMMIT_OFFSET;
// send msg
......@@ -1313,17 +1314,6 @@ int32_t tmq_seek(tmq_t* tmq, const tmq_topic_vgroup_t* offset) {
#endif
SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) {
/*int64_t reqOffset;*/
/*if (pVg->currentOffset >= 0) {*/
/*reqOffset = pVg->currentOffset;*/
/*} else {*/
/*if (tmq->resetOffsetCfg == TMQ_CONF__RESET_OFFSET__NONE) {*/
/*tscError("unable to poll since no committed offset but reset offset is set to none");*/
/*return NULL;*/
/*}*/
/*reqOffset = tmq->resetOffsetCfg;*/
/*}*/
SMqPollReq* pReq = taosMemoryCalloc(1, sizeof(SMqPollReq));
if (pReq == NULL) {
return NULL;
......
source/common/src/tmsg.c
浏览文件 @ 322b8c30
......@@ -305,7 +305,7 @@ static int32_t tDeserializeSClientHbReq(SDecoder *pDecoder, SClientHbReq *pReq)
taosArrayPush(desc.subDesc, &sDesc);
}
}
ASSERT(desc.subPlanNum == taosArrayGetSize(desc.subDesc));
taosArrayPush(pReq->query->queryDesc, &desc);
......@@ -5770,3 +5770,40 @@ int32_t tDecodeSMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
}
return 0;
}
int32_t tEncodeSSingleDeleteReq(SEncoder *pEncoder, const SSingleDeleteReq *pReq) {
if (tEncodeI64(pEncoder, pReq->uid) < 0) return -1;
if (tEncodeI64(pEncoder, pReq->ts) < 0) return -1;
return 0;
}
int32_t tDecodeSSingleDeleteReq(SDecoder *pDecoder, SSingleDeleteReq *pReq) {
if (tDecodeI64(pDecoder, &pReq->uid) < 0) return -1;
if (tDecodeI64(pDecoder, &pReq->ts) < 0) return -1;
return 0;
}
int32_t tEncodeSBatchDeleteReq(SEncoder *pEncoder, const SBatchDeleteReq *pReq) {
if (tEncodeI64(pEncoder, pReq->suid) < 0) return -1;
int32_t sz = taosArrayGetSize(pReq->deleteReqs);
if (tEncodeI32(pEncoder, sz) < 0) return -1;
for (int32_t i = 0; i < sz; i++) {
SSingleDeleteReq *pOneReq = taosArrayGet(pReq->deleteReqs, i);
if (tEncodeSSingleDeleteReq(pEncoder, pOneReq) < 0) return -1;
}
return 0;
}
int32_t tDecodeSBatchDeleteReq(SDecoder *pDecoder, SBatchDeleteReq *pReq) {
if (tDecodeI64(pDecoder, &pReq->suid) < 0) return -1;
int32_t sz;
if (tDecodeI32(pDecoder, &sz) < 0) return -1;
pReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
if (pReq->deleteReqs == NULL) return -1;
for (int32_t i = 0; i < sz; i++) {
SSingleDeleteReq deleteReq;
if (tDecodeSSingleDeleteReq(pDecoder, &deleteReq) < 0) return -1;
taosArrayPush(pReq->deleteReqs, &deleteReq);
}
return 0;
}
source/common/src/trow.c
浏览文件 @ 322b8c30
......@@ -32,9 +32,13 @@ const uint8_t tdVTypeByte[2][3] = {{
};
// declaration
static uint8_t tdGetBitmapByte(uint8_t byte);
static int32_t tdCompareColId(const void *arg1, const void *arg2);
static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2);
static uint8_t tdGetBitmapByte(uint8_t byte);
static bool tdSTSRowIterGetTpVal(STSRowIter *pIter, col_type_t colType, int32_t offset, SCellVal *pVal);
static bool tdSTSRowIterGetKvVal(STSRowIter *pIter, col_id_t colId, col_id_t *nIdx, SCellVal *pVal);
static bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t flen, uint32_t offset,
col_id_t colIdx, SCellVal *pVal);
static bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVal);
static void tdSCellValPrint(SCellVal *pVal, int8_t colType);
// implementation
/**
......@@ -330,14 +334,14 @@ void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag) {
tdSTSRowIterInit(&iter, pSchema);
tdSTSRowIterReset(&iter, row);
printf("%s >>>type:%d,sver:%d ", tag, (int32_t)TD_ROW_TYPE(row), (int32_t)TD_ROW_SVER(row));
for (int i = 0; i < pSchema->numOfCols; ++i) {
STColumn *stCol = pSchema->columns + i;
SCellVal sVal = {255, NULL};
if (!tdSTSRowIterNext(&iter, stCol->colId, stCol->type, &sVal)) {
STColumn *cols = (STColumn *)&iter.pSchema->columns;
while (true) {
SCellVal sVal = {.valType = 255, NULL};
if (!tdSTSRowIterNext(&iter, &sVal)) {
break;
}
ASSERT(sVal.valType == 0 || sVal.valType == 1 || sVal.valType == 2);
tdSCellValPrint(&sVal, stCol->type);
tdSCellValPrint(&sVal, cols[iter.colIdx - 1].type);
}
printf("\n");
}
......@@ -420,6 +424,16 @@ void tdSCellValPrint(SCellVal *pVal, int8_t colType) {
}
}
static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2) {
if (*(col_id_t *)key1 > ((SKvRowIdx *)key2)->colId) {
return 1;
} else if (*(col_id_t *)key1 < ((SKvRowIdx *)key2)->colId) {
return -1;
} else {
return 0;
}
}
bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVal) {
if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
tdRowSetVal(pVal, TD_VTYPE_NORM, TD_ROW_KEY_ADDR(pRow));
......@@ -456,7 +470,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl
return true;
}
bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal) {
bool tdSTSRowIterFetch(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal) {
if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
pVal->val = &pIter->pRow->ts;
pVal->valType = TD_VTYPE_NORM;
......@@ -477,10 +491,10 @@ bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCe
return false;
}
}
tdGetTpRowDataOfCol(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
++pIter->colIdx;
} else if (TD_IS_KV_ROW(pIter->pRow)) {
return tdGetKvRowValOfColEx(pIter, colId, colType, &pIter->kvIdx, pVal);
return tdSTSRowIterGetKvVal(pIter, colId, &pIter->kvIdx, pVal);
} else {
pVal->valType = TD_VTYPE_NONE;
terrno = TSDB_CODE_INVALID_PARA;
......@@ -489,13 +503,68 @@ bool tdSTSRowIterNext(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCe
return true;
}
bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType, col_id_t *nIdx, SCellVal *pVal) {
bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) {
if (pIter->colIdx >= pIter->pSchema->numOfCols) {
return false;
}
STColumn *pCol = &pIter->pSchema->columns[pIter->colIdx];
if (pCol->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
pVal->val = &pIter->pRow->ts;
pVal->valType = TD_VTYPE_NORM;
++pIter->colIdx;
return true;
}
if (TD_IS_TP_ROW(pIter->pRow)) {
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
} else if (TD_IS_KV_ROW(pIter->pRow)) {
tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
} else {
ASSERT(0);
}
++pIter->colIdx;
return true;
}
bool tdSTSRowIterGetTpVal(STSRowIter *pIter, col_type_t colType, int32_t offset, SCellVal *pVal) {
STSRow *pRow = pIter->pRow;
if (pRow->statis == 0) {
pVal->valType = TD_VTYPE_NORM;
if (IS_VAR_DATA_TYPE(colType)) {
pVal->val = POINTER_SHIFT(pRow, *(VarDataOffsetT *)POINTER_SHIFT(TD_ROW_DATA(pRow), offset));
} else {
pVal->val = POINTER_SHIFT(TD_ROW_DATA(pRow), offset);
}
return TSDB_CODE_SUCCESS;
}
if (tdGetBitmapValType(pIter->pBitmap, pIter->colIdx - 1, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
pVal->valType = TD_VTYPE_NONE;
return terrno;
}
if (pVal->valType == TD_VTYPE_NORM) {
if (IS_VAR_DATA_TYPE(colType)) {
pVal->val = POINTER_SHIFT(pRow, *(VarDataOffsetT *)POINTER_SHIFT(TD_ROW_DATA(pRow), offset));
} else {
pVal->val = POINTER_SHIFT(TD_ROW_DATA(pRow), offset);
}
}
return true;
}
bool tdSTSRowIterGetKvVal(STSRowIter *pIter, col_id_t colId, col_id_t *nIdx, SCellVal *pVal) {
STSRow *pRow = pIter->pRow;
SKvRowIdx *pKvIdx = NULL;
bool colFound = false;
col_id_t kvNCols = tdRowGetNCols(pRow) - 1;
void *pColIdx = TD_ROW_COL_IDX(pRow);
while (*nIdx < kvNCols) {
pKvIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(pRow), *nIdx * sizeof(SKvRowIdx));
pKvIdx = (SKvRowIdx *)POINTER_SHIFT(pColIdx, *nIdx * sizeof(SKvRowIdx));
if (pKvIdx->colId == colId) {
++(*nIdx);
pVal->val = POINTER_SHIFT(pRow, pKvIdx->offset);
......@@ -518,48 +587,13 @@ bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType,
}
}
#ifdef TD_SUPPORT_BITMAP
int16_t colIdx = -1;
if (pKvIdx) colIdx = POINTER_DISTANCE(pKvIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx);
if (tdGetBitmapValType(pIter->pBitmap, colIdx, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
pVal->valType = TD_VTYPE_NONE;
}
#else
pVal->valType = isNull(pVal->val, colType) ? TD_VTYPE_NULL : TD_VTYPE_NORM;
#endif
return true;
}
bool tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colType, int32_t offset, SCellVal *pVal) {
STSRow *pRow = pIter->pRow;
if (IS_VAR_DATA_TYPE(colType)) {
pVal->val = POINTER_SHIFT(pRow, *(VarDataOffsetT *)POINTER_SHIFT(TD_ROW_DATA(pRow), offset));
} else {
pVal->val = POINTER_SHIFT(TD_ROW_DATA(pRow), offset);
}
#ifdef TD_SUPPORT_BITMAP
if (tdGetBitmapValType(pIter->pBitmap, pIter->colIdx - 1, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
if (tdGetBitmapValType(pIter->pBitmap, pIter->kvIdx - 1, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
pVal->valType = TD_VTYPE_NONE;
}
#else
pVal->valType = isNull(pVal->val, colType) ? TD_VTYPE_NULL : TD_VTYPE_NORM;
#endif
return true;
}
static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2) {
if (*(col_id_t *)key1 > ((SKvRowIdx *)key2)->colId) {
return 1;
} else if (*(col_id_t *)key1 < ((SKvRowIdx *)key2)->colId) {
return -1;
} else {
return 0;
}
}
int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow) {
STColumn *pTColumn;
SColVal *pColVal;
......@@ -625,7 +659,7 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow) {
if (maxVarDataLen > 0) {
varBuf = taosMemoryMalloc(maxVarDataLen);
if (!varBuf) {
if(isAlloc) {
if (isAlloc) {
taosMemoryFreeClear(*ppRow);
}
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -673,6 +707,19 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow) {
return 0;
}
static FORCE_INLINE int32_t tdCompareColId(const void *arg1, const void *arg2) {
int32_t colId = *(int32_t *)arg1;
STColumn *pCol = (STColumn *)arg2;
if (colId < pCol->colId) {
return -1;
} else if (colId == pCol->colId) {
return 0;
} else {
return 1;
}
}
bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal) {
if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
pVal->val = &pIter->pRow->ts;
......@@ -712,19 +759,6 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
return true;
}
static int32_t tdCompareColId(const void *arg1, const void *arg2) {
int32_t colId = *(int32_t *)arg1;
STColumn *pCol = (STColumn *)arg2;
if (colId < pCol->colId) {
return -1;
} else if (colId == pCol->colId) {
return 0;
} else {
return 1;
}
}
int32_t tdGetBitmapValTypeII(const void *pBitmap, int16_t colIdx, TDRowValT *pValType) {
if (!pBitmap || colIdx < 0) {
TASSERT(0);
......@@ -938,7 +972,7 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
break;
case TD_VTYPE_NONE:
if (!pBuilder->hasNone) pBuilder->hasNone = true;
break;
return TSDB_CODE_SUCCESS;
default:
ASSERT(0);
break;
......@@ -970,13 +1004,11 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
STSRow *row = pBuilder->pBuf;
// No need to store None/Null values.
SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset);
pColIdx->colId = colId;
pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN
if (valType == TD_VTYPE_NORM) {
// ts key stored in STSRow.ts
SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset);
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
pColIdx->colId = colId;
pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
if (IS_VAR_DATA_TYPE(colType)) {
if (isCopyVarData) {
memcpy(ptr, val, varDataTLen(val));
......@@ -987,26 +1019,6 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
TD_ROW_LEN(row) += TYPE_BYTES[colType];
}
}
#ifdef TD_SUPPORT_BACK2
// NULL/None value
else {
SKvRowIdx *pColIdx = (SKvRowIdx *)POINTER_SHIFT(TD_ROW_COL_IDX(row), offset);
char *ptr = (char *)POINTER_SHIFT(row, TD_ROW_LEN(row));
pColIdx->colId = colId;
pColIdx->offset = TD_ROW_LEN(row); // the offset include the TD_ROW_HEAD_LEN
const void *nullVal = getNullValue(colType);
if (IS_VAR_DATA_TYPE(colType)) {
if (isCopyVarData) {
memcpy(ptr, nullVal, varDataTLen(nullVal));
}
TD_ROW_LEN(row) += varDataTLen(nullVal);
} else {
memcpy(ptr, nullVal, TYPE_BYTES[colType]);
TD_ROW_LEN(row) += TYPE_BYTES[colType];
}
}
#endif
return 0;
}
......@@ -1044,24 +1056,6 @@ int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
memcpy(POINTER_SHIFT(TD_ROW_DATA(row), offset), val, TYPE_BYTES[colType]);
}
}
#ifdef TD_SUPPORT_BACK2
// NULL/None value
else {
// TODO: Null value for new data types imported since 3.0 need to be defined.
const void *nullVal = getNullValue(colType);
if (IS_VAR_DATA_TYPE(colType)) {
// ts key stored in STSRow.ts
*(VarDataOffsetT *)POINTER_SHIFT(TD_ROW_DATA(row), offset) = TD_ROW_LEN(row);
if (isCopyVarData) {
memcpy(POINTER_SHIFT(row, TD_ROW_LEN(row)), nullVal, varDataTLen(nullVal));
}
TD_ROW_LEN(row) += varDataTLen(nullVal);
} else {
memcpy(POINTER_SHIFT(TD_ROW_DATA(row), offset), nullVal, TYPE_BYTES[colType]);
}
}
#endif
return 0;
}
......@@ -1329,7 +1323,7 @@ void tdSTSRowIterReset(STSRowIter *pIter, STSRow *pRow) {
pIter->pRow = pRow;
pIter->pBitmap = tdGetBitmapAddr(pRow, pRow->type, pIter->pSchema->flen, tdRowGetNCols(pRow));
pIter->offset = 0;
pIter->colIdx = PRIMARYKEY_TIMESTAMP_COL_ID;
pIter->colIdx = 0; // PRIMARYKEY_TIMESTAMP_COL_ID;
pIter->kvIdx = 0;
}
......@@ -1367,4 +1361,4 @@ void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColV
*pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value);
}
}
}
\ No newline at end of file
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ 322b8c30
......@@ -364,6 +364,7 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_VND_CHECK_ALTER_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
......
source/dnode/mnode/impl/inc/mndTopic.h
浏览文件 @ 322b8c30
......@@ -32,6 +32,7 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic);
SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw);
int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
int32_t mndCheckTopicExist(SMnode *pMnode, SDbObj *pDb);
const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]);
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 322b8c30
......@@ -995,11 +995,13 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
mDebug("trans:%d, used to drop db:%s", pTrans->id, pDb->name);
mndTransSetDbName(pTrans, pDb->name, NULL);
if (mndCheckTopicExist(pMnode, pDb) < 0) goto _OVER;
if (mndSetDropDbRedoLogs(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndSetDropDbCommitLogs(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropOffsetByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropSubByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
/*if (mndDropOffsetByDB(pMnode, pTrans, pDb) != 0) goto _OVER;*/
/*if (mndDropSubByDB(pMnode, pTrans, pDb) != 0) goto _OVER;*/
/*if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;*/
if (mndDropStreamByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropSmasByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
......@@ -1706,7 +1708,7 @@ static void setPerfSchemaDbCfg(SDbObj *pDbObj) {
static bool mndGetTablesOfDbFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
SVgObj *pVgroup = pObj;
int32_t *numOfTables = p1;
int64_t uid = *(int64_t*)p2;
int64_t uid = *(int64_t *)p2;
if (pVgroup->dbUid == uid) {
*numOfTables += pVgroup->numOfTables;
}
......
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @ 322b8c30
......@@ -837,7 +837,7 @@ int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
sdbCancelFetch(pSdb, pIter);
mError("db:%s, failed to drop stream:%s since sourceDbUid:%" PRId64 " not match with targetDbUid:%" PRId64,
pDb->name, pStream->name, pStream->sourceDbUid, pStream->targetDbUid);
terrno = TSDB_CODE_MND_STREAM_ALREADY_EXIST;
terrno = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
return -1;
} else {
#if 0
......@@ -929,14 +929,31 @@ static int32_t mndRetrieveStream(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->status, true);
char sourceDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->sourceDb, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&n, varDataVal(sourceDB));
varDataSetLen(sourceDB, strlen(varDataVal(sourceDB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->sourceDb, true);
colDataAppend(pColInfo, numOfRows, (const char *)&sourceDB, false);
char targetDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->targetDb, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&n, varDataVal(targetDB));
varDataSetLen(targetDB, strlen(varDataVal(targetDB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetDb, true);
colDataAppend(pColInfo, numOfRows, (const char *)&targetDB, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetSTbName, true);
if (pStream->targetSTbName[0] == 0) {
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, NULL, true);
} else {
char targetSTB[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pStream->targetSTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
strcpy(&targetSTB[VARSTR_HEADER_SIZE], tNameGetTableName(&n));
varDataSetLen(targetSTB, strlen(varDataVal(targetSTB)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&targetSTB, false);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->watermark, false);
......
source/dnode/mnode/impl/src/mndSubscribe.c
浏览文件 @ 322b8c30
......@@ -398,13 +398,27 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
}
}
// 8. TODO generate logs
mInfo("rebalance calculation completed, rebalanced vg:");
// 8. generate logs
mInfo("mq rebalance: calculation completed, rebalanced vg:");
for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {
SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);
mInfo("vgId:%d, moved from consumer:%" PRId64 ", to consumer:%" PRId64, pOutputRebVg->pVgEp->vgId,
mInfo("mq rebalance: vgId:%d, moved from consumer:%" PRId64 ", to consumer:%" PRId64, pOutputRebVg->pVgEp->vgId,
pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);
}
{
void *pIter = NULL;
while (1) {
pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
if (pIter == NULL) break;
SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
mInfo("mq rebalance: final cfg: consumer %ld has %d vg", pConsumerEp->consumerId, sz);
for (int32_t i = 0; i < sz; i++) {
SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
mInfo("mq rebalance: final cfg: vg %d to consumer %ld", pVgEp->vgId, pConsumerEp->consumerId);
}
}
}
// 9. clear
taosHashCleanup(pHash);
......
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @ 322b8c30
......@@ -782,6 +782,26 @@ static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) {
sdbCancelFetch(pSdb, pIter);
}
int32_t mndCheckTopicExist(SMnode *pMnode, SDbObj *pDb) {
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
SMqTopicObj *pTopic = NULL;
while (1) {
pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
if (pIter == NULL) break;
if (pTopic->dbUid == pDb->uid) {
sdbRelease(pSdb, pTopic);
terrno = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
return -1;
}
sdbRelease(pSdb, pTopic);
}
return 0;
}
int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
int32_t code = 0;
SSdb *pSdb = pMnode->pSdb;
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 322b8c30
......@@ -172,7 +172,7 @@ int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
int32_t tsdbGetStbIdList(SMeta* pMeta, int64_t suid, SArray* list);
SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid,
const char* stbFullName, int32_t vgId);
const char* stbFullName, int32_t vgId, SBatchDeleteReq* pDeleteReq);
// sma
int32_t smaInit();
......
source/dnode/vnode/src/sma/smaTimeRange.c
浏览文件 @ 322b8c30
......@@ -119,7 +119,7 @@ int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
SName stbFullName = {0};
tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
SVCreateStbReq pReq = {0};
pReq.name = (char*)tNameGetTableName(&stbFullName);
pReq.name = (char *)tNameGetTableName(&stbFullName);
pReq.suid = pCfg->dstTbUid;
pReq.schemaRow = pCfg->schemaRow;
pReq.schemaTag = pCfg->schemaTag;
......@@ -200,8 +200,9 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
goto _err;
}
SBatchDeleteReq deleteReq;
SSubmitReq *pSubmitReq = tdBlockToSubmit((const SArray *)msg, pTsmaStat->pTSchema, true, pTsmaStat->pTSma->dstTbUid,
pTsmaStat->pTSma->dstTbName, pTsmaStat->pTSma->dstVgId);
pTsmaStat->pTSma->dstTbName, pTsmaStat->pTSma->dstVgId, &deleteReq);
if (!pSubmitReq) {
smaError("vgId:%d, failed to gen submit blk while tsma insert for smaIndex %" PRIi64 " since %s", SMA_VID(pSma),
......@@ -230,4 +231,4 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
_err:
tdUnRefSmaStat(pSma, pStat);
return TSDB_CODE_FAILED;
}
\ No newline at end of file
}
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 322b8c30
......@@ -325,7 +325,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
for (int32_t i = 0; i < colActual; i++) {
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
SCellVal sVal = {0};
if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
break;
}
if (colDataAppend(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
......
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @ 322b8c30
......@@ -16,7 +16,7 @@
#include "tq.h"
SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, bool createTb, int64_t suid,
const char* stbFullName, int32_t vgId) {
const char* stbFullName, int32_t vgId, SBatchDeleteReq* deleteReq) {
SSubmitReq* ret = NULL;
SArray* schemaReqs = NULL;
SArray* schemaReqSz = NULL;
......@@ -33,10 +33,13 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
schemaReqSz = taosArrayInit(sz, sizeof(int32_t));
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
STagVal tagVal = {
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
if (pDataBlock->info.type == STREAM_DELETE_DATA) {
//
}
STagVal tagVal = {
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
};
STag* pTag = NULL;
taosArrayClear(tagArray);
......@@ -177,17 +180,45 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
}
void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
const SArray* pRes = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
const SArray* pRes = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
SBatchDeleteReq deleteReq = {0};
tqDebug("vgId:%d, task %d write into table, block num: %d", TD_VID(pVnode), pTask->taskId, (int32_t)pRes->size);
ASSERT(pTask->tbSink.pTSchema);
deleteReq.deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
SSubmitReq* pReq = tdBlockToSubmit(pRes, pTask->tbSink.pTSchema, true, pTask->tbSink.stbUid,
pTask->tbSink.stbFullName, pVnode->config.vgId);
pTask->tbSink.stbFullName, pVnode->config.vgId, &deleteReq);
tqDebug("vgId:%d, task %d convert blocks over, put into write-queue", TD_VID(pVnode), pTask->taskId);
int32_t code;
int32_t len;
tEncodeSize(tEncodeSBatchDeleteReq, &deleteReq, len, code);
if (code < 0) {
//
ASSERT(0);
}
SEncoder encoder;
void* buf = taosMemoryCalloc(1, len + sizeof(SMsgHead));
void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
tEncoderInit(&encoder, abuf, len);
tEncodeSBatchDeleteReq(&encoder, &deleteReq);
tEncoderClear(&encoder);
if (taosArrayGetSize(deleteReq.deleteReqs) != 0) {
SRpcMsg msg = {
.msgType = TDMT_VND_BATCH_DEL,
.pCont = buf,
.contLen = len + sizeof(SMsgHead),
};
if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
tqDebug("failed to put into write-queue since %s", terrstr());
}
}
taosArrayDestroy(deleteReq.deleteReqs);
/*tPrintFixedSchemaSubmitReq(pReq, pTask->tbSink.pTSchema);*/
// build write msg
SRpcMsg msg = {
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 322b8c30
......@@ -182,7 +182,7 @@ static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
if (IS_VAR_DATA_TYPE(pCol->info.type)) {
pSupInfo->buildBuf[i] = taosMemoryMalloc(pCol->info.bytes);
tsdbInfo("-------------------%d\n", pCol->info.bytes);
// tsdbInfo("-------------------%d\n", pCol->info.bytes);
}
}
......
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 322b8c30
......@@ -29,6 +29,7 @@ static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void
static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t code = 0;
......@@ -190,6 +191,9 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
case TDMT_VND_DELETE:
if (vnodeProcessDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
break;
case TDMT_VND_BATCH_DEL:
if (vnodeProcessBatchDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
break;
/* TQ */
case TDMT_VND_MQ_VG_CHANGE:
if (tqProcessVgChangeReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
......@@ -1053,6 +1057,23 @@ static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void
return 0;
}
static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
SBatchDeleteReq deleteReq;
SDecoder decoder;
tDecoderInit(&decoder, pReq, len);
tDecodeSBatchDeleteReq(&decoder, &deleteReq);
int32_t sz = taosArrayGetSize(deleteReq.deleteReqs);
for (int32_t i = 0; i < sz; i++) {
SSingleDeleteReq *pOneReq = taosArrayGet(deleteReq.deleteReqs, i);
int32_t code = tsdbDeleteTableData(pVnode->pTsdb, version, deleteReq.suid, pOneReq->uid, pOneReq->ts, pOneReq->ts);
if (code) {
// TODO
}
}
return 0;
}
static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
int32_t code = 0;
SDecoder *pCoder = &(SDecoder){0};
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 322b8c30
......@@ -741,6 +741,7 @@ typedef struct STimeSliceOperatorInfo {
SArray* pPrevRow; // SArray<SGroupValue>
SArray* pNextRow; // SArray<SGroupValue>
SArray* pLinearInfo; // SArray<SFillLinearInfo>
bool fillLastPoint;
bool isPrevRowSet;
bool isNextRowSet;
int32_t fillType; // fill type
......@@ -1017,6 +1018,7 @@ bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs,
TSKEY* pEndTs, int32_t rows, int32_t start, int64_t gap, SHashObj* pStDeleted);
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup);
bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid);
......
source/libs/executor/inc/tfill.h
浏览文件 @ 322b8c30
......@@ -37,7 +37,6 @@ typedef struct SFillLinearInfo {
SPoint start;
SPoint end;
bool hasNull;
bool fillLastPoint;
int16_t type;
int32_t bytes;
} SFillLinearInfo;
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 322b8c30
......@@ -1277,8 +1277,12 @@ void destroyTableQueryInfoImpl(STableQueryInfo* pTableQueryInfo) {
}
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset) {
bool init = false;
for (int32_t i = 0; i < numOfOutput; ++i) {
pCtx[i].resultInfo = getResultEntryInfo(pResult, i, rowEntryInfoOffset);
if (init) {
continue;
}
struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
if (isRowEntryCompleted(pResInfo) && isRowEntryInitialized(pResInfo)) {
......@@ -1295,6 +1299,8 @@ void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numO
} else {
pResInfo->initialized = true;
}
} else {
init = true;
}
}
}
......@@ -3416,6 +3422,7 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
}
int32_t code = createDiskbasedBuf(&pAggSup->pResultBuf, defaultPgsz, defaultBufsz, pKey, tsTempDir);
if (code != TSDB_CODE_SUCCESS) {
qError("Create agg result buf failed since %s", tstrerror(code));
return code;
}
......@@ -3435,7 +3442,11 @@ int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInf
return code;
}
doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
code = doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
for (int32_t i = 0; i < numOfCols; ++i) {
pSup->pCtx[i].pBuf = pAggSup->pResultBuf;
}
......
source/libs/executor/src/joinoperator.c
浏览文件 @ 322b8c30
......@@ -132,6 +132,8 @@ void destroyMergeJoinOperator(void* param, int32_t numOfOutput) {
SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param;
nodesDestroyNode(pJoinOperator->pCondAfterMerge);
pJoinOperator->pRes = blockDataDestroy(pJoinOperator->pRes);
taosMemoryFreeClear(param);
}
......
source/libs/executor/src/scanoperator.c
浏览文件 @ 322b8c30
......@@ -319,6 +319,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
return TSDB_CODE_SUCCESS;
} else {
qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo));
*status = FUNC_DATA_REQUIRED_DATA_LOAD;
}
}
......@@ -326,20 +327,19 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
ASSERT(*status == FUNC_DATA_REQUIRED_DATA_LOAD);
// try to filter data block according to sma info
if (pTableScanInfo->pFilterNode != NULL) {
if (!loadSMA) {
doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);
}
bool keep = doFilterByBlockSMA(pTableScanInfo->pFilterNode, pBlock->pBlockAgg, taosArrayGetSize(pBlock->pDataBlock),
pBlockInfo->rows);
if (!keep) {
qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->filterOutBlocks += 1;
(*status) = FUNC_DATA_REQUIRED_FILTEROUT;
return TSDB_CODE_SUCCESS;
if (pTableScanInfo->pFilterNode != NULL && (!loadSMA)) {
bool success = doLoadBlockSMA(pTableScanInfo, pBlock, pTaskInfo);
if (success) {
size_t size = taosArrayGetSize(pBlock->pDataBlock);
bool keep = doFilterByBlockSMA(pTableScanInfo->pFilterNode, pBlock->pBlockAgg, size, pBlockInfo->rows);
if (!keep) {
qDebug("%s data block filter out by block SMA, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->filterOutBlocks += 1;
(*status) = FUNC_DATA_REQUIRED_FILTEROUT;
return TSDB_CODE_SUCCESS;
}
}
}
......@@ -1174,10 +1174,15 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1;
STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);
bool isClosed = false;
STimeWindow win = {.skey = INT64_MIN, .ekey = INT64_MAX};
if (isOverdue(tsCol[rowId], &pInfo->twAggSup)) {
win = getActiveTimeWindow(NULL, &dumyInfo, tsCol[rowId], &pInfo->interval, TSDB_ORDER_ASC);
isClosed = isCloseWindow(&win, &pInfo->twAggSup);
}
// must check update info first.
bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]);
if ((update || (isSignleIntervalWindow(pInfo) && isCloseWindow(&win, &pInfo->twAggSup) &&
if ((update || (isSignleIntervalWindow(pInfo) && isClosed &&
isDeletedWindow(&win, pBlock->info.groupId, pInfo->sessionSup.pIntervalAggSup))) && out) {
appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid);
}
......@@ -2458,81 +2463,6 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
#if 0
int32_t maxNumOfTables = (int32_t)pResultInfo->capacity;
STagScanInfo *pInfo = pOperator->info;
SSDataBlock *pRes = pInfo->pRes;
int32_t count = 0;
SArray* pa = GET_TABLEGROUP(pRuntimeEnv, 0);
int32_t functionId = getExprFunctionId(&pOperator->exprSupp.pExprInfo[0]);
if (functionId == FUNCTION_TID_TAG) { // return the tags & table Id
assert(pQueryAttr->numOfOutput == 1);
SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0];
int32_t rsize = pExprInfo->base.resSchema.bytes;
count = 0;
int16_t bytes = pExprInfo->base.resSchema.bytes;
int16_t type = pExprInfo->base.resSchema.type;
for(int32_t i = 0; i < pQueryAttr->numOfTags; ++i) {
if (pQueryAttr->tagColList[i].colId == pExprInfo->base.pColumns->info.colId) {
bytes = pQueryAttr->tagColList[i].bytes;
type = pQueryAttr->tagColList[i].type;
break;
}
}
SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
while(pInfo->curPos < pInfo->totalTables && count < maxNumOfTables) {
int32_t i = pInfo->curPos++;
STableQueryInfo *item = taosArrayGetP(pa, i);
char *output = pColInfo->pData + count * rsize;
varDataSetLen(output, rsize - VARSTR_HEADER_SIZE);
output = varDataVal(output);
STableId* id = TSDB_TABLEID(item->pTable);
*(int16_t *)output = 0;
output += sizeof(int16_t);
*(int64_t *)output = id->uid; // memory align problem, todo serialize
output += sizeof(id->uid);
*(int32_t *)output = id->tid;
output += sizeof(id->tid);
*(int32_t *)output = pQueryAttr->vgId;
output += sizeof(pQueryAttr->vgId);
char* data = NULL;
if (pExprInfo->base.pColumns->info.colId == TSDB_TBNAME_COLUMN_INDEX) {
data = tsdbGetTableName(item->pTable);
} else {
data = tsdbGetTableTagVal(item->pTable, pExprInfo->base.pColumns->info.colId, type, bytes);
}
doSetTagValueToResultBuf(output, data, type, bytes);
count += 1;
}
//qDebug("QInfo:0x%"PRIx64" create (tableId, tag) info completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
} else if (functionId == FUNCTION_COUNT) {// handle the "count(tbname)" query
SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
*(int64_t*)pColInfo->pData = pInfo->totalTables;
count = 1;
pOperator->status = OP_EXEC_DONE;
//qDebug("QInfo:0x%"PRIx64" create count(tbname) query, res:%d rows:1", GET_TASKID(pRuntimeEnv), count);
} else { // return only the tags|table name etc.
#endif
STagScanInfo* pInfo = pOperator->info;
SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0];
SSDataBlock* pRes = pInfo->pRes;
......@@ -2610,6 +2540,8 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
STagScanInfo* pInfo = (STagScanInfo*)param;
pInfo->pRes = blockDataDestroy(pInfo->pRes);
taosArrayDestroy(pInfo->pColMatchInfo);
taosMemoryFreeClear(param);
}
......
source/libs/executor/src/tsimplehash.c
浏览文件 @ 322b8c30
......@@ -13,38 +13,37 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "tsimplehash.h"
#include "os.h"
#include "taoserror.h"
#define SHASH_DEFAULT_LOAD_FACTOR 0.75
#define HASH_MAX_CAPACITY (1024*1024*16)
#define HASH_MAX_CAPACITY (1024 * 1024 * 16)
#define SHASH_NEED_RESIZE(_h) ((_h)->size >= (_h)->capacity * SHASH_DEFAULT_LOAD_FACTOR)
#define GET_SHASH_NODE_KEY(_n, _dl) ((char*)(_n) + sizeof(SHNode) + (_dl))
#define GET_SHASH_NODE_DATA(_n) ((char*)(_n) + sizeof(SHNode))
#define GET_SHASH_NODE_KEY(_n, _dl) ((char *)(_n) + sizeof(SHNode) + (_dl))
#define GET_SHASH_NODE_DATA(_n) ((char *)(_n) + sizeof(SHNode))
#define HASH_INDEX(v, c) ((v) & ((c)-1))
#define HASH_NEED_RESIZE(_h) ((_h)->size >= (_h)->capacity * SHASH_DEFAULT_LOAD_FACTOR)
#define HASH_INDEX(v, c) ((v) & ((c)-1))
#define FREE_HASH_NODE(_n) \
do { \
taosMemoryFreeClear(_n); \
#define FREE_HASH_NODE(_n) \
do { \
taosMemoryFreeClear(_n); \
} while (0);
typedef struct SHNode {
struct SHNode *next;
char data[];
struct SHNode *next;
char data[];
} SHNode;
struct SSHashObj {
SHNode **hashList;
size_t capacity; // number of slots
int64_t size; // number of elements in hash table
_hash_fn_t hashFp; // hash function
_equal_fn_t equalFp; // equal function
int32_t keyLen;
int32_t dataLen;
SHNode **hashList;
size_t capacity; // number of slots
int64_t size; // number of elements in hash table
_hash_fn_t hashFp; // hash function
_equal_fn_t equalFp; // equal function
int32_t keyLen;
int32_t dataLen;
};
static FORCE_INLINE int32_t taosHashCapacity(int32_t length) {
......@@ -62,7 +61,7 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn, size_t keyLen, size_t
capacity = 4;
}
SSHashObj* pHashObj = (SSHashObj*) taosMemoryCalloc(1, sizeof(SSHashObj));
SSHashObj *pHashObj = (SSHashObj *)taosMemoryCalloc(1, sizeof(SSHashObj));
if (pHashObj == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
......@@ -72,7 +71,7 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn, size_t keyLen, size_t
pHashObj->capacity = taosHashCapacity((int32_t)capacity);
pHashObj->equalFp = memcmp;
pHashObj->hashFp = fn;
pHashObj->hashFp = fn;
ASSERT((pHashObj->capacity & (pHashObj->capacity - 1)) == 0);
pHashObj->keyLen = keyLen;
......@@ -91,7 +90,7 @@ int32_t tSimpleHashGetSize(const SSHashObj *pHashObj) {
if (pHashObj == NULL) {
return 0;
}
return (int32_t)atomic_load_64((int64_t*)&pHashObj->size);
return (int32_t)atomic_load_64((int64_t *)&pHashObj->size);
}
static SHNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, size_t dsize, uint32_t hashVal) {
......@@ -108,41 +107,42 @@ static SHNode *doCreateHashNode(const void *key, size_t keyLen, const void *pDat
}
static void taosHashTableResize(SSHashObj *pHashObj) {
if (!HASH_NEED_RESIZE(pHashObj)) {
if (!SHASH_NEED_RESIZE(pHashObj)) {
return;
}
int32_t newCapacity = (int32_t)(pHashObj->capacity << 1u);
if (newCapacity > HASH_MAX_CAPACITY) {
// uDebug("current capacity:%zu, maximum capacity:%d, no resize applied due to limitation is reached",
// pHashObj->capacity, HASH_MAX_CAPACITY);
// uDebug("current capacity:%zu, maximum capacity:%d, no resize applied due to limitation is reached",
// pHashObj->capacity, HASH_MAX_CAPACITY);
return;
}
int64_t st = taosGetTimestampUs();
void *pNewEntryList = taosMemoryRealloc(pHashObj->hashList, sizeof(void *) * newCapacity);
void *pNewEntryList = taosMemoryRealloc(pHashObj->hashList, sizeof(void *) * newCapacity);
if (pNewEntryList == NULL) {
// qWarn("hash resize failed due to out of memory, capacity remain:%zu", pHashObj->capacity);
// qWarn("hash resize failed due to out of memory, capacity remain:%zu", pHashObj->capacity);
return;
}
size_t inc = newCapacity - pHashObj->capacity;
memset((char*)pNewEntryList + pHashObj->capacity * sizeof(void*), 0, inc);
memset((char *)pNewEntryList + pHashObj->capacity * sizeof(void *), 0, inc);
pHashObj->hashList = pNewEntryList;
pHashObj->capacity = newCapacity;
for (int32_t idx = 0; idx < pHashObj->capacity; ++idx) {
SHNode* pNode = pHashObj->hashList[idx];
SHNode *pNext;
SHNode *pPrev = NULL;
SHNode *pNode = pHashObj->hashList[idx];
if (pNode == NULL) {
continue;
}
SHNode *pNext;
SHNode *pPrev = NULL;
while (pNode != NULL) {
void* key = GET_SHASH_NODE_KEY(pNode, pHashObj->dataLen);
void *key = GET_SHASH_NODE_KEY(pNode, pHashObj->dataLen);
uint32_t hashVal = (*pHashObj->hashFp)(key, (uint32_t)pHashObj->dataLen);
int32_t newIdx = HASH_INDEX(hashVal, pHashObj->capacity);
......@@ -166,8 +166,9 @@ static void taosHashTableResize(SSHashObj *pHashObj) {
int64_t et = taosGetTimestampUs();
// uDebug("hash table resize completed, new capacity:%d, load factor:%f, elapsed time:%fms", (int32_t)pHashObj->capacity,
// ((double)pHashObj->size) / pHashObj->capacity, (et - st) / 1000.0);
// uDebug("hash table resize completed, new capacity:%d, load factor:%f, elapsed time:%fms",
// (int32_t)pHashObj->capacity,
// ((double)pHashObj->size) / pHashObj->capacity, (et - st) / 1000.0);
}
int32_t tSimpleHashPut(SSHashObj *pHashObj, const void *key, const void *data) {
......@@ -210,7 +211,7 @@ int32_t tSimpleHashPut(SSHashObj *pHashObj, const void *key, const void *data) {
pNewNode->next = pHashObj->hashList[slot];
pHashObj->hashList[slot] = pNewNode;
atomic_add_fetch_64(&pHashObj->size, 1);
} else { //update data
} else { // update data
memcpy(GET_SHASH_NODE_DATA(pNode), data, pHashObj->dataLen);
}
......@@ -230,9 +231,7 @@ static FORCE_INLINE SHNode *doSearchInEntryList(SSHashObj *pHashObj, const void
return pNode;
}
static FORCE_INLINE bool taosHashTableEmpty(const SSHashObj *pHashObj) {
return tSimpleHashGetSize(pHashObj) == 0;
}
static FORCE_INLINE bool taosHashTableEmpty(const SSHashObj *pHashObj) { return tSimpleHashGetSize(pHashObj) == 0; }
void *tSimpleHashGet(SSHashObj *pHashObj, const void *key) {
if (pHashObj == NULL || taosHashTableEmpty(pHashObj) || key == NULL) {
......@@ -299,9 +298,9 @@ size_t tSimpleHashGetMemSize(const SSHashObj *pHashObj) {
return (pHashObj->capacity * sizeof(void *)) + sizeof(SHNode) * tSimpleHashGetSize(pHashObj) + sizeof(SSHashObj);
}
void *tSimpleHashGetKey(const SSHashObj* pHashObj, void *data, size_t* keyLen) {
void *tSimpleHashGetKey(const SSHashObj *pHashObj, void *data, size_t *keyLen) {
int32_t offset = offsetof(SHNode, data);
SHNode *node = ((SHNode*)(char*)data - offset);
SHNode *node = ((SHNode *)(char *)data - offset);
if (keyLen != NULL) {
*keyLen = pHashObj->keyLen;
}
......
source/libs/function/inc/taggfunction.h → source/libs/function/inc/tfunctionInt.h
浏览文件 @ 322b8c30
......@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TAGGFUNCTION_H
#define TDENGINE_TAGGFUNCTION_H
#ifndef TDENGINE_TFUNCTIONINT_H
#define TDENGINE_TFUNCTIONINT_H
#ifdef __cplusplus
extern "C" {
......@@ -28,17 +28,6 @@ extern "C" {
#include "function.h"
#include "tudf.h"
#define AVG_FUNCTION_INTER_BUFFER_SIZE 50
#define DATA_SET_FLAG ',' // to denote the output area has data, not null value
#define DATA_SET_FLAG_SIZE sizeof(DATA_SET_FLAG)
typedef struct SInterpInfoDetail {
TSKEY ts; // interp specified timestamp
int8_t type;
int8_t primaryCol;
} SInterpInfoDetail;
bool topbot_datablock_filter(SqlFunctionCtx *pCtx, const char *minval, const char *maxval);
/**
......@@ -57,4 +46,4 @@ static FORCE_INLINE void initResultRowEntry(SResultRowEntryInfo *pResInfo, int32
}
#endif
#endif // TDENGINE_TAGGFUNCTION_H
#endif // TDENGINE_TFUNCTIONINT_H
source/libs/function/src/builtinsimpl.c
浏览文件 @ 322b8c30
......@@ -18,10 +18,10 @@
#include "function.h"
#include "query.h"
#include "querynodes.h"
#include "taggfunction.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "tdigest.h"
#include "tfunctionInt.h"
#include "tglobal.h"
#include "thistogram.h"
#include "tpercentile.h"
......@@ -312,14 +312,6 @@ typedef struct SGroupKeyInfo {
#define GET_TS_LIST(x) ((TSKEY*)((x)->ptsList))
#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])
#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx) \
do { \
for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
SqlFunctionCtx* __ctx = (ctx)->tagInfo.pTagCtxList[_i]; \
__ctx->fpSet.process(__ctx); \
} \
} while (0);
#define DO_UPDATE_SUBSID_RES(ctx, ts) \
do { \
for (int32_t _i = 0; _i < (ctx)->subsidiaries.num; ++_i) { \
......@@ -506,8 +498,7 @@ int32_t functionFinalizeWithResultBuf(SqlFunctionCtx* pCtx, SSDataBlock* pBlock,
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0;
cleanupResultRowEntry(pResInfo);
pResInfo->isNullRes = (pResInfo->isNullRes == 1) ? 1 : (pResInfo->numOfRes == 0);;
char* in = finalResult;
colDataAppend(pCol, pBlock->info.rows, in, pResInfo->isNullRes);
......@@ -4284,9 +4275,9 @@ static int32_t histogramFunctionImpl(SqlFunctionCtx* pCtx, bool isPartial) {
}
if (!isPartial) {
SET_VAL(GET_RES_INFO(pCtx), numOfElems, pInfo->numOfBins);
GET_RES_INFO(pCtx)->numOfRes = pInfo->numOfBins;
} else {
SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
GET_RES_INFO(pCtx)->numOfRes = 1;
}
return TSDB_CODE_SUCCESS;
}
......
source/libs/function/src/taggfunction.c 已删除 100644 → 0
浏览文件 @ 253257e4
此差异已折叠。
source/libs/function/src/texpr.c 已删除 100644 → 0
浏览文件 @ 253257e4
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "function.h"
#include "os.h"
#include "texception.h"
#include "tmsg.h"
static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *));
void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *)) {
if (pNode == NULL) {
return;
}
if (pNode->nodeType == TEXPR_BINARYEXPR_NODE || pNode->nodeType == TEXPR_UNARYEXPR_NODE) {
doExprTreeDestroy(&pNode, fp);
}
taosMemoryFree(pNode);
}
static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *)) {
if (*pExpr == NULL) {
return;
}
taosMemoryFree(*pExpr);
*pExpr = NULL;
}
// TODO: these three functions should be made global
static void* exception_calloc(size_t nmemb, size_t size) {
void* p = taosMemoryCalloc(nmemb, size);
if (p == NULL) {
THROW(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
return p;
}
static void* exception_malloc(size_t size) {
void* p = taosMemoryMalloc(size);
if (p == NULL) {
THROW(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
return p;
}
static UNUSED_FUNC char* exception_strdup(const char* str) {
char* p = strdup(str);
if (p == NULL) {
THROW(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
return p;
}
source/libs/function/src/tfunctionInt.c 0 → 100644
浏览文件 @ 322b8c30
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "taosdef.h"
#include "tmsg.h"
#include "thash.h"
#include "ttypes.h"
#include "function.h"
#include "tbuffer.h"
#include "tcompression.h"
#include "tdatablock.h"
#include "tfunctionInt.h"
#include "thistogram.h"
#include "tpercentile.h"
#include "ttszip.h"
#include "tudf.h"
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell) {
pCell->initialized = false;
}
int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock) {
int32_t maxRows = 0;
for (int32_t j = 0; j < num; ++j) {
#if 0
int32_t id = pCtx[j].functionId;
/*
* ts, tag, tagprj function can not decide the output number of current query
* the number of output result is decided by main output
*/
if (id == FUNCTION_TS || id == FUNCTION_TAG || id == FUNCTION_TAGPRJ) {
continue;
}
#endif
SResultRowEntryInfo *pResInfo = GET_RES_INFO(&pCtx[j]);
if (pResInfo != NULL && maxRows < pResInfo->numOfRes) {
maxRows = pResInfo->numOfRes;
}
}
assert(maxRows >= 0);
blockDataEnsureCapacity(pResBlock, maxRows);
for(int32_t i = 0; i < num; ++i) {
SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, i);
SResultRowEntryInfo *pResInfo = GET_RES_INFO(&pCtx[i]);
if (pResInfo->numOfRes == 0) {
for(int32_t j = 0; j < pResInfo->numOfRes; ++j) {
colDataAppend(pCol, j, NULL, true); // TODO add set null data api
}
} else {
for (int32_t j = 0; j < pResInfo->numOfRes; ++j) {
colDataAppend(pCol, j, GET_ROWCELL_INTERBUF(pResInfo), false);
}
}
}
pResBlock->info.rows = maxRows;
return maxRows;
}
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry) {
assert(pEntry != NULL);
return pEntry->complete;
}
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry) {
return pEntry->initialized;
}
source/libs/parser/src/parInsert.c
浏览文件 @ 322b8c30
......@@ -1192,7 +1192,10 @@ static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks,
pBuilder->hasNone = true;
}
tdSRowEnd(pBuilder);
*gotRow = true;
#ifdef TD_DEBUG_PRINT_ROW
STSchema* pSTSchema = tdGetSTSChemaFromSSChema(schema, spd->numOfCols, 1);
tdSRowPrint(row, pSTSchema, __func__);
......@@ -1201,7 +1204,6 @@ static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks,
}
// *len = pBuilder->extendedRowSize;
tdSRowEnd(pBuilder);
return TSDB_CODE_SUCCESS;
}
......
source/libs/stream/src/streamUpdate.c
浏览文件 @ 322b8c30
......@@ -124,7 +124,7 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma
}
pInfo->numBuckets = DEFAULT_BUCKET_SIZE;
pInfo->pCloseWinSBF = NULL;
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);
pInfo->pMap = taosHashInit(DEFAULT_MAP_CAPACITY, hashFn, true, HASH_NO_LOCK);
pInfo->maxVersion = 0;
pInfo->scanGroupId = 0;
......
source/libs/sync/inc/syncInt.h
浏览文件 @ 322b8c30
......@@ -163,6 +163,7 @@ typedef struct SSyncNode {
bool changing;
int64_t startTime;
int64_t leaderTime;
int64_t lastReplicateTime;
} SSyncNode;
......
source/libs/sync/inc/syncRespMgr.h
浏览文件 @ 322b8c30
......@@ -32,9 +32,9 @@ typedef struct SRespStub {
} SRespStub;
typedef struct SSyncRespMgr {
SHashObj * pRespHash;
SHashObj *pRespHash;
int64_t ttl;
void * data;
void *data;
TdThreadMutex mutex;
uint64_t seqNum;
} SSyncRespMgr;
......@@ -46,7 +46,8 @@ int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t index);
int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub);
int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub);
void syncRespClean(SSyncRespMgr *pObj);
void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl);
void syncRespCleanRsp(SSyncRespMgr *pObj);
void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp);
#ifdef __cplusplus
}
......
source/libs/sync/src/syncMain.c
浏览文件 @ 322b8c30
......@@ -1100,6 +1100,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
int64_t timeNow = taosGetTimestampMs();
pSyncNode->startTime = timeNow;
pSyncNode->leaderTime = timeNow;
pSyncNode->lastReplicateTime = timeNow;
syncNodeEventLog(pSyncNode, "sync open");
......@@ -2015,6 +2016,8 @@ void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term) {
}
}
void syncNodeLeaderChangeRsp(SSyncNode* pSyncNode) { syncRespCleanRsp(pSyncNode->pSyncRespMgr); }
void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// maybe clear leader cache
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
......@@ -2028,6 +2031,9 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// reset elect timer
syncNodeResetElectTimer(pSyncNode);
// send rsp to client
syncNodeLeaderChangeRsp(pSyncNode);
// call back
if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeFollowerCb != NULL) {
pSyncNode->pFsm->FpBecomeFollowerCb(pSyncNode->pFsm);
......@@ -2068,6 +2074,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
//
void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
pSyncNode->leaderTime = taosGetTimestampMs();
// reset restoreFinish
pSyncNode->restoreFinish = false;
......@@ -2954,8 +2962,11 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
}
ths->restoreFinish = true;
int64_t restoreDelay = taosGetTimestampMs() - ths->leaderTime;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "restore finish, index:%" PRId64, pEntry->index);
snprintf(eventLog, sizeof(eventLog), "restore finish, index:%ld, elapsed:%ld ms, ", pEntry->index,
restoreDelay);
syncNodeEventLog(ths, eventLog);
}
}
......
source/libs/sync/src/syncRespMgr.c
浏览文件 @ 322b8c30
......@@ -108,13 +108,19 @@ int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStu
return 0; // get none object
}
void syncRespCleanRsp(SSyncRespMgr *pObj) {
taosThreadMutexLock(&(pObj->mutex));
syncRespCleanByTTL(pObj, -1, true);
taosThreadMutexUnlock(&(pObj->mutex));
}
void syncRespClean(SSyncRespMgr *pObj) {
taosThreadMutexLock(&(pObj->mutex));
syncRespCleanByTTL(pObj, pObj->ttl);
syncRespCleanByTTL(pObj, pObj->ttl, false);
taosThreadMutexUnlock(&(pObj->mutex));
}
void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL);
int cnt = 0;
int sum = 0;
......@@ -126,12 +132,12 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
while (pStub) {
size_t len;
void * key = taosHashGetKey(pStub, &len);
void *key = taosHashGetKey(pStub, &len);
uint64_t *pSeqNum = (uint64_t *)key;
sum++;
int64_t nowMS = taosGetTimestampMs();
if (nowMS - pStub->createTime > ttl) {
if (nowMS - pStub->createTime > ttl || -1 == ttl) {
taosArrayPush(delIndexArray, pSeqNum);
cnt++;
......@@ -148,7 +154,14 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
pStub->rpcMsg.pCont = NULL;
pStub->rpcMsg.contLen = 0;
pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta);
// TODO: and make rpcMsg body, call commit cb
// pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta);
pStub->rpcMsg.code = TSDB_CODE_SYN_NOT_LEADER;
if (pStub->rpcMsg.info.handle != NULL) {
tmsgSendRsp(&(pStub->rpcMsg));
}
}
pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub);
......
source/util/src/terror.c
浏览文件 @ 322b8c30
......@@ -281,11 +281,12 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC, "Invalid topic")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_QUERY, "Topic with invalid query")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_OPTION, "Topic with invalid option")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST, "Consumer not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED, "Consumer unchanged")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED, "Topic unchanged")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST, "Subcribe not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_OFFSET_NOT_EXIST, "Offset not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_SUBSCRIBED, "Topic subscribed cannot be dropped")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_MUST_BE_DELETED, "Topic must be dropped first")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CGROUP_USED, "Consumer group being used by some consumer")
// mnode-stream
......
tools/CMakeLists.txt
浏览文件 @ 322b8c30
......@@ -112,7 +112,7 @@ ELSE ()
COMMAND CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib go build -a -ldflags "-s -w -X github.com/taosdata/taosadapter/version.Version=${taos_version} -X github.com/taosdata/taosadapter/version.CommitID=${taosadapter_commit_sha1}"
COMMAND CGO_CFLAGS=-I${CMAKE_CURRENT_SOURCE_DIR}/../include/client CGO_LDFLAGS=-L${CMAKE_BINARY_DIR}/build/lib go build -a -o taosadapter-debug -ldflags "-X github.com/taosdata/taosadapter/version.Version=${taos_version} -X github.com/taosdata/taosadapter/version.CommitID=${taosadapter_commit_sha1}"
INSTALL_COMMAND
COMMAND wget -c https://github.com/upx/upx/releases/download/v3.96/upx-3.96-${PLATFORM_ARCH_STR}_linux.tar.xz -O $ENV{HOME}/upx.tar.xz && tar -xvJf $ENV{HOME}/upx.tar.xz -C $ENV{HOME}/ --strip-components 1 > /dev/null && $ENV{HOME}/upx taosadapter || :
COMMAND wget -nc https://github.com/upx/upx/releases/download/v3.96/upx-3.96-${PLATFORM_ARCH_STR}_linux.tar.xz -O $ENV{HOME}/upx.tar.xz && tar -xvJf $ENV{HOME}/upx.tar.xz -C $ENV{HOME}/ --strip-components 1 > /dev/null && $ENV{HOME}/upx taosadapter || :
COMMAND cmake -E copy taosadapter ${CMAKE_BINARY_DIR}/build/bin
COMMAND cmake -E make_directory ${CMAKE_BINARY_DIR}/test/cfg/
COMMAND cmake -E copy ./example/config/taosadapter.toml ${CMAKE_BINARY_DIR}/test/cfg/
......
tools/shell/src/shellArguments.c
浏览文件 @ 322b8c30
......@@ -38,7 +38,7 @@
#define SHELL_STARTUP "Check the details of the service status."
#define SHELL_WIDTH "Set the default binary display width, default is 30."
#define SHELL_NET_ROLE "Net role when network connectivity test, options: client|server."
#define SHELL_PKG_LEN "Packet length used for net test, default is 1024 bytes."
#define SHELL_PKT_LEN "Packet length used for net test, default is 1024 bytes."
#define SHELL_PKT_NUM "Packet numbers used for net test, default is 100."
#define SHELL_VERSION "Print program version."
#define SHELL_EMAIL "<support@taosdata.com>"
......@@ -62,7 +62,7 @@ void shellPrintHelp() {
printf("%s%s%s%s\r\n", indent, "-f,", indent, SHELL_FILE);
printf("%s%s%s%s\r\n", indent, "-h,", indent, SHELL_HOST);
printf("%s%s%s%s\r\n", indent, "-k,", indent, SHELL_CHECK);
printf("%s%s%s%s\r\n", indent, "-l,", indent, SHELL_PKG_LEN);
printf("%s%s%s%s\r\n", indent, "-l,", indent, SHELL_PKT_LEN);
printf("%s%s%s%s\r\n", indent, "-n,", indent, SHELL_NET_ROLE);
printf("%s%s%s%s\r\n", indent, "-N,", indent, SHELL_PKT_NUM);
printf("%s%s%s%s\r\n", indent, "-p,", indent, SHELL_PASSWORD);
......@@ -105,11 +105,11 @@ static struct argp_option shellOptions[] = {
{"startup", 't', 0, 0, SHELL_STARTUP},
{"display-width", 'w', "WIDTH", 0, SHELL_WIDTH},
{"netrole", 'n', "NETROLE", 0, SHELL_NET_ROLE},
{"pktlen", 'l', "PKTLEN", 0, SHELL_PKG_LEN},
{"pktlen", 'l', "PKTLEN", 0, SHELL_PKT_LEN},
#ifdef WEBSOCKET
{"dsn", 'E', "DSN", 0, SHELL_DSN},
{"restful", 'R', 0, 0, SHELL_REST},
{"timeout", 'T', "SECONDS", 0, SHELL_TIMEOUT},
{"timeout", 'T', "SECONDS", 0, SHELL_TIMEOUT},
#endif
{"pktnum", 'N', "PKTNUM", 0, SHELL_PKT_NUM},
{0},
......@@ -228,7 +228,7 @@ int32_t shellParseArgsWithoutArgp(int argc, char *argv[]) {
SShellArgs *pArgs = &shell.args;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "--usage") == 0 || strcmp(argv[i], "-?") == 0) {
if (strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "--usage") == 0 || strcmp(argv[i], "-?") == 0 || strcmp(argv[i], "/?") == 0) {
shellParseSingleOpt('?', NULL);
return 0;
}
......
taos-tools @ 3c7dafee
Subproject commit 3c7dafeea3e558968165b73bee0f51024898e3da
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或