Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 b9a248d7 编写于 作者: H Hongze Cheng
浏览文件
操作

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

上级 ce9afbae e4e6277c
展开全部 隐藏空白更改
内联 并排
Showing with 226 addition and 114 deletion
cmake/taostools_CMakeLists.txt.in
浏览文件 @ b9a248d7
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
# taos-tools # taos-tools
ExternalProject_Add(taos-tools ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG 285b5e0 GIT_TAG 8207c74
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR "" BINARY_DIR ""
#BUILD_IN_SOURCE TRUE #BUILD_IN_SOURCE TRUE
......
cmake/taosws_CMakeLists.txt.in
浏览文件 @ b9a248d7
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
# taosws-rs # taosws-rs
ExternalProject_Add(taosws-rs ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG 76bc64d GIT_TAG 1bdfca3
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs" SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR "" BINARY_DIR ""
#BUILD_IN_SOURCE TRUE #BUILD_IN_SOURCE TRUE
......
docs/en/12-taos-sql/01-data-type.md
浏览文件 @ b9a248d7
此差异已折叠。
docs/en/12-taos-sql/06-select.md
浏览文件 @ b9a248d7
...@@ -68,7 +68,7 @@ A query can be performed on some or all columns. Data and tag columns can all be ...@@ -68,7 +68,7 @@ A query can be performed on some or all columns. Data and tag columns can all be
### Wildcards ### Wildcards
You can use an asterisk (\*) as a wildcard character to indicate all columns. For standard tables, the asterisk indicates only data columns. For supertables and subtables, tag columns are also included. You can use an asterisk (\*) as a wildcard character to indicate all columns. For normal tables or sub-tables, the asterisk indicates only data columns. For supertables, tag columns are also included when using asterisk (\*).
```sql ```sql
SELECT * FROM d1001; SELECT * FROM d1001;
......
docs/zh/12-taos-sql/01-data-type.md
浏览文件 @ b9a248d7
--- ---
sidebar_label: 数据类型 sidebar_label: 数据类型
title: 数据类型 title: 数据类型
description: "TDengine 支持的数据类型: 时间戳、浮点型、JSON 类型等" description: 'TDengine 支持的数据类型: 时间戳、浮点型、JSON 类型等'
--- ---
## 时间戳 ## 时间戳
...@@ -9,64 +9,65 @@ description: "TDengine 支持的数据类型: 时间戳、浮点型、JSON 类 ...@@ -9,64 +9,65 @@ description: "TDengine 支持的数据类型: 时间戳、浮点型、JSON 类
使用 TDengine，最重要的是时间戳。创建并插入记录、查询历史记录的时候，均需要指定时间戳。时间戳有如下规则： 使用 TDengine，最重要的是时间戳。创建并插入记录、查询历史记录的时候，均需要指定时间戳。时间戳有如下规则：
- 时间格式为 `YYYY-MM-DD HH:mm:ss.MS`，默认时间分辨率为毫秒。比如：`2017-08-12 18:25:58.128` - 时间格式为 `YYYY-MM-DD HH:mm:ss.MS`，默认时间分辨率为毫秒。比如：`2017-08-12 18:25:58.128`
- 内部函数 now 是客户端的当前时间 - 内部函数 NOW 是客户端的当前时间
- 插入记录时，如果时间戳为 now，插入数据时使用提交这条记录的客户端的当前时间 - 插入记录时，如果时间戳为 NOW，插入数据时使用提交这条记录的客户端的当前时间
- Epoch Time：时间戳也可以是一个长整数，表示从 UTC 时间 1970-01-01 00:00:00 开始的毫秒数。相应地，如果所在 Database 的时间精度设置为“微秒”，则长整型格式的时间戳含义也就对应于从 UTC 时间 1970-01-01 00:00:00 开始的微秒数；纳秒精度逻辑类似。 - Epoch Time：时间戳也可以是一个长整数，表示从 UTC 时间 1970-01-01 00:00:00 开始的毫秒数。相应地，如果所在 Database 的时间精度设置为“微秒”，则长整型格式的时间戳含义也就对应于从 UTC 时间 1970-01-01 00:00:00 开始的微秒数；纳秒精度逻辑相同。
- 时间可以加减，比如 now-2h，表明查询时刻向前推 2 个小时（最近 2 小时）。数字后面的时间单位可以是 b(纳秒)、u(微秒)、a(毫秒)、s(秒)、m(分)、h(小时)、d(天)、w(周)。 比如 `select * from t1 where ts > now-2w and ts <= now-1w`，表示查询两周前整整一周的数据。在指定降采样操作（down sampling）的时间窗口（interval）时，时间单位还可以使用 n (自然月) 和 y (自然年)。 - 时间可以加减，比如 NOW-2h，表明查询时刻向前推 2 个小时（最近 2 小时）。数字后面的时间单位可以是 b（纳秒）、u（微秒）、a（毫秒）、s（秒）、m（分）、h（小时）、d（天）、w（周）。 比如 `SELECT * FROM t1 WHERE ts > NOW-2w AND ts <= NOW-1w`，表示查询两周前整整一周的数据。在指定降采样操作（Down Sampling）的时间窗口（Interval）时，时间单位还可以使用 n（自然月）和 y（自然年）。
TDengine 缺省的时间戳精度是毫秒，但通过在 `CREATE DATABASE` 时传递的 PRECISION 参数也可以支持微秒和纳秒。 TDengine 缺省的时间戳精度是毫秒，但通过在 `CREATE DATABASE` 时传递的 `PRECISION` 参数也可以支持微秒和纳秒。
```sql ```sql
CREATE DATABASE db_name PRECISION 'ns'; CREATE DATABASE db_name PRECISION 'ns';
``` ```
## 数据类型 ## 数据类型
在 TDengine 中，普通表的数据模型中可使用以下数据类型。 在 TDengine 中，普通表的数据模型中可使用以下数据类型。
| # | **类型** | **Bytes** | **说明** | | # | **类型** | **Bytes** | **说明** |
| --- | :-------: | --------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | --- | :---------------: | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| 1 | TIMESTAMP | 8 | 时间戳。缺省精度毫秒，可支持微秒和纳秒，详细说明见上节。 | | 1 | TIMESTAMP | 8 | 时间戳。缺省精度毫秒，可支持微秒和纳秒，详细说明见上节。 |
| 2 | INT | 4 | 整型，范围 [-2^31, 2^31-1] | | 2 | INT | 4 | 整型，范围 [-2^31, 2^31-1] |
| 3 | INT UNSIGNED| 4| 无符号整数，[0, 2^32-1] | 3 | INT UNSIGNED | 4 | 无符号整数，[0, 2^32-1] |
| 4 | BIGINT | 8 | 长整型，范围 [-2^63, 2^63-1] | | 4 | BIGINT | 8 | 长整型，范围 [-2^63, 2^63-1] |
| 5 | BIGINT UNSIGNED | 8 | 长整型，范围 [0, 2^64-1] | | 5 | BIGINT UNSIGNED | 8 | 长整型，范围 [0, 2^64-1] |
| 6 | FLOAT | 4 | 浮点型，有效位数 6-7，范围 [-3.4E38, 3.4E38] | | 6 | FLOAT | 4 | 浮点型，有效位数 6-7，范围 [-3.4E38, 3.4E38] |
| 7 | DOUBLE | 8 | 双精度浮点型，有效位数 15-16，范围 [-1.7E308, 1.7E308] | | 7 | DOUBLE | 8 | 双精度浮点型，有效位数 15-16，范围 [-1.7E308, 1.7E308] |
| 8 | BINARY | 自定义 | 记录单字节字符串，建议只用于处理 ASCII 可见字符，中文等多字节字符需使用 nchar。 | | 8 | BINARY | 自定义 | 记录单字节字符串，建议只用于处理 ASCII 可见字符，中文等多字节字符需使用 NCHAR |
| 9 | SMALLINT | 2 | 短整型， 范围 [-32768, 32767] | | 9 | SMALLINT | 2 | 短整型， 范围 [-32768, 32767] |
| 10 | SMALLINT UNSIGNED | 2| 无符号短整型，范围 [0, 65535] | | 10 | SMALLINT UNSIGNED | 2 | 无符号短整型，范围 [0, 65535] |
| 11 | TINYINT | 1 | 单字节整型，范围 [-128, 127] | | 11 | TINYINT | 1 | 单字节整型，范围 [-128, 127] |
| 12 | TINYINT UNSIGNED | 1 | 无符号单字节整型，范围 [0, 255] | | 12 | TINYINT UNSIGNED | 1 | 无符号单字节整型，范围 [0, 255] |
| 13 | BOOL | 1 | 布尔型，{true, false} | | 13 | BOOL | 1 | 布尔型，{true, false} |
| 14 | NCHAR | 自定义 | 记录包含多字节字符在内的字符串，如中文字符。每个 nchar 字符占用 4 bytes 的存储空间。字符串两端使用单引号引用，字符串内的单引号需用转义字符 `\’`。nchar 使用时须指定字符串大小，类型为 nchar(10) 的列表示此列的字符串最多存储 10 个 nchar 字符，会固定占用 40 bytes 的空间。如果用户字符串长度超出声明长度，将会报错。 | | 14 | NCHAR | 自定义 | 记录包含多字节字符在内的字符串，如中文字符。每个 NCHAR 字符占用 4 字节的存储空间。字符串两端使用单引号引用，字符串内的单引号需用转义字符 `\'`。NCHAR 使用时须指定字符串大小，类型为 NCHAR(10) 的列表示此列的字符串最多存储 10 个 NCHAR 字符，会固定占用 40 字节的空间。如果用户字符串长度超出声明长度，将会报错。 |
| 15 | JSON | | json 数据类型， 只有 tag 可以是 json 格式 | | 15 | JSON | | JSON 数据类型， 只有 Tag 可以是 JSON 格式 |
| 16 | VARCHAR | 自定义 | BINARY类型的别名 | | 16 | VARCHAR | 自定义 | BINARY 类型的别名 |
:::note :::note
- TDengine 对 SQL 语句中的英文字符不区分大小写，自动转化为小写执行。因此用户大小写敏感的字符串及密码，需要使用单引号将字符串引起来。
- 虽然 BINARY 类型在底层存储上支持字节型的二进制字符，但不同编程语言对二进制数据的处理方式并不保证一致，因此建议在 BINARY 类型中只存储 ASCII 可见字符，而避免存储不可见字符。多字节的数据，例如中文字符，则需要使用 NCHAR 类型进行保存。如果强行使用 BINARY 类型保存中文字符，虽然有时也能正常读写，但并不带有字符集信息，很容易出现数据乱码甚至数据损坏等情况。 - 虽然 BINARY 类型在底层存储上支持字节型的二进制字符，但不同编程语言对二进制数据的处理方式并不保证一致，因此建议在 BINARY 类型中只存储 ASCII 可见字符，而避免存储不可见字符。多字节的数据，例如中文字符，则需要使用 NCHAR 类型进行保存。如果强行使用 BINARY 类型保存中文字符，虽然有时也能正常读写，但并不带有字符集信息，很容易出现数据乱码甚至数据损坏等情况。
- BINARY 类型理论上最长可以有 16374 字节。binary 仅支持字符串输入，字符串两端需使用单引号引用。使用时须指定大小，如 binary(20) 定义了最长为 20 个单字节字符的字符串，每个字符占 1 byte 的存储空间，总共固定占用 20 bytes 的空间，此时如果用户字符串超出 20 字节将会报错。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示，即 `\’`。 - BINARY 类型理论上最长可以有 16,374 字节。BINARY 仅支持字符串输入，字符串两端需使用单引号引用。使用时须指定大小，如 BINARY(20) 定义了最长为 20 个单字节字符的字符串，每个字符占 1 字节的存储空间，总共固定占用 20 字节的空间，此时如果用户字符串超出 20 字节将会报错。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示，即 `\'`。
- SQL 语句中的数值类型将依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型，因此在使用时要注意相应类型越界的情况。例如，9999999999999999999 会认为超过长整型的上边界而溢出，而 9999999999999999999.0 会被认为是有效的浮点数。 - SQL 语句中的数值类型将依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型，因此在使用时要注意相应类型越界的情况。例如，9999999999999999999 会认为超过长整型的上边界而溢出，而 9999999999999999999.0 会被认为是有效的浮点数。
::: :::
## 常量 ## 常量
TDengine支持多个类型的常量，细节如下表：
TDengine 支持多个类型的常量，细节如下表：
| # | **语法** | **类型** | **说明** |
| --- | :-------: | --------- | -------------------------------------- | | # | **语法** | **类型** | **说明** |
| 1 | [{+ \| -}]123 | BIGINT | 整型数值的字面量的类型均为BIGINT。如果用户输入超过了BIGINT的表示范围，TDengine 按BIGINT对数值进行截断。| | --- | :-----------------------------------------------: | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| 2 | 123.45 | DOUBLE | 浮点数值的字面量的类型均为DOUBLE。TDengine依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型。| | 1 | [{+ \| -}]123 | BIGINT | 整型数值的字面量的类型均为 BIGINT。如果用户输入超过了 BIGINT 的表示范围，TDengine 按 BIGINT 对数值进行截断。 |
| 3 | 1.2E3 | DOUBLE | 科学计数法的字面量的类型为DOUBLE。| | 2 | 123.45 | DOUBLE | 浮点数值的字面量的类型均为 DOUBLE。TDengine 依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型。 |
| 4 | 'abc' | BINARY | 单引号括住的内容为字符串字面值，其类型为BINARY，BINARY的size为实际的字符个数。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示，即 \'。| | 3 | 1.2E3 | DOUBLE | 科学计数法的字面量的类型为 DOUBLE。 |
| 5 | "abc" | BINARY | 双引号括住的内容为字符串字面值，其类型为BINARY，BINARY的size为实际的字符个数。对于字符串内的双引号，可以用转义字符反斜线加单引号来表示，即 \"。 | | 4 | 'abc' | BINARY | 单引号括住的内容为字符串字面值，其类型为 BINARY，BINARY 的 Size 为实际的字符个数。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示，即 `\'`。 |
| 6 | TIMESTAMP {'literal' \| "literal"} | TIMESTAMP | TIMESTAMP关键字表示后面的字符串字面量需要被解释为TIMESTAMP类型。字符串需要满足YYYY-MM-DD HH:mm:ss.MS格式，其时间分辨率为当前数据库的时间分辨率。 | | 5 | "abc" | BINARY | 双引号括住的内容为字符串字面值，其类型为 BINARY，BINARY 的 Size 为实际的字符个数。对于字符串内的双引号，可以用转义字符反斜线加单引号来表示，即 `\"`。 |
| 7 | {TRUE \| FALSE} | BOOL | 布尔类型字面量。 | | 6 | TIMESTAMP {'literal' \| "literal"} | TIMESTAMP | TIMESTAMP 关键字表示后面的字符串字面量需要被解释为 TIMESTAMP 类型。字符串需要满足 YYYY-MM-DD HH:mm:ss.MS 格式，其时间分辨率为当前数据库的时间分辨率。 |
| 8 | {'' \| "" \| '\t' \| "\t" \| ' ' \| " " \| NULL } | -- | 空值字面量。可以用于任意类型。| | 7 | {TRUE \| FALSE} | BOOL | 布尔类型字面量。 |
| 8 | {'' \| "" \| '\t' \| "\t" \| ' ' \| " " \| NULL } | -- | 空值字面量。可以用于任意类型。 |
:::note :::note
- TDengine依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型，因此在使用时要注意相应类型越界的情况。例如，9999999999999999999会认为超过长整型的上边界而溢出，而9999999999999999999.0会被认为是有效的浮点数。
- TDengine 依据是否存在小数点，或使用科学计数法表示，来判断数值类型是否为整型或者浮点型，因此在使用时要注意相应类型越界的情况。例如，9999999999999999999 会认为超过长整型的上边界而溢出，而 9999999999999999999.0 会被认为是有效的浮点数。
::: :::
docs/zh/12-taos-sql/06-select.md
浏览文件 @ b9a248d7
...@@ -69,7 +69,7 @@ order_expr: ...@@ -69,7 +69,7 @@ order_expr:
### 通配符 ### 通配符
通配符 \* 可以用于代指全部列。对于普通表，结果中只有普通列。对于超级表和子表，还包含了 TAG 列。 通配符 \* 可以用于代指全部列。对于普通表和子表，结果中只有普通列。对于超级表，还包含了 TAG 列。
```sql ```sql
SELECT * FROM d1001; SELECT * FROM d1001;
......
include/common/tcommon.h
浏览文件 @ b9a248d7
...@@ -116,6 +116,7 @@ enum { ...@@ -116,6 +116,7 @@ enum {
STREAM_INPUT__DATA_RETRIEVE, STREAM_INPUT__DATA_RETRIEVE,
STREAM_INPUT__GET_RES, STREAM_INPUT__GET_RES,
STREAM_INPUT__CHECKPOINT, STREAM_INPUT__CHECKPOINT,
STREAM_INPUT__REF_DATA_BLOCK,
STREAM_INPUT__DESTROY, STREAM_INPUT__DESTROY,
}; };
......
include/libs/stream/tstream.h
浏览文件 @ b9a248d7
...@@ -125,6 +125,14 @@ typedef struct { ...@@ -125,6 +125,14 @@ typedef struct {
SArray* blocks; // SArray<SSDataBlock> SArray* blocks; // SArray<SSDataBlock>
} SStreamDataBlock; } SStreamDataBlock;
// ref data block, for delete
typedef struct {
int8_t type;
int64_t ver;
int32_t* dataRef;
SSDataBlock* pBlock;
} SStreamRefDataBlock;
typedef struct { typedef struct {
int8_t type; int8_t type;
} SStreamCheckpoint; } SStreamCheckpoint;
...@@ -339,7 +347,8 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem ...@@ -339,7 +347,8 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
qDebug("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data); qDebug("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data);
taosWriteQitem(pTask->inputQueue->queue, pSubmitClone); taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
// qStreamInput(pTask->exec.executor, pSubmitClone); // qStreamInput(pTask->exec.executor, pSubmitClone);
} else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) { } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE ||
pItem->type == STREAM_INPUT__REF_DATA_BLOCK) {
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
// qStreamInput(pTask->exec.executor, pItem); // qStreamInput(pTask->exec.executor, pItem);
} else if (pItem->type == STREAM_INPUT__CHECKPOINT) { } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
...@@ -492,7 +501,9 @@ typedef struct { ...@@ -492,7 +501,9 @@ typedef struct {
int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq); int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq);
int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq); int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq);
void tFreeStreamDispatchReq(SStreamDispatchReq* pReq); void tDeleteStreamRetrieveReq(SStreamRetrieveReq* pReq);
void tDeleteStreamDispatchReq(SStreamDispatchReq* pReq);
int32_t streamSetupTrigger(SStreamTask* pTask); int32_t streamSetupTrigger(SStreamTask* pTask);
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ b9a248d7
...@@ -59,7 +59,7 @@ static void destroySTqHandle(void* data) { ...@@ -59,7 +59,7 @@ static void destroySTqHandle(void* data) {
tqCloseReader(pData->execHandle.pExecReader); tqCloseReader(pData->execHandle.pExecReader);
walCloseReader(pData->pWalReader); walCloseReader(pData->pWalReader);
taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid); taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid);
} else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE){ } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
walCloseReader(pData->pWalReader); walCloseReader(pData->pWalReader);
tqCloseReader(pData->execHandle.pExecReader); tqCloseReader(pData->execHandle.pExecReader);
} }
...@@ -664,7 +664,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLe ...@@ -664,7 +664,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLe
tqError("vgId:%d, build new consumer handle %s for consumer %d, but old consumerId is %ld", req.vgId, req.subKey, tqError("vgId:%d, build new consumer handle %s for consumer %d, but old consumerId is %ld", req.vgId, req.subKey,
req.newConsumerId, req.oldConsumerId); req.newConsumerId, req.oldConsumerId);
} }
ASSERT(req.newConsumerId != -1); if (req.newConsumerId == -1) {
tqError("vgId:%d, tq invalid rebalance request, new consumerId %ld", req.vgId, req.newConsumerId);
return 0;
}
STqHandle tqHandle = {0}; STqHandle tqHandle = {0};
pHandle = &tqHandle; pHandle = &tqHandle;
/*taosInitRWLatch(&pExec->lock);*/ /*taosInitRWLatch(&pExec->lock);*/
...@@ -876,6 +879,9 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) { ...@@ -876,6 +879,9 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
taosArrayDestroy(pRes->uidList); taosArrayDestroy(pRes->uidList);
int32_t* pRef = taosMemoryMalloc(sizeof(int32_t));
*pRef = 1;
void* pIter = NULL; void* pIter = NULL;
while (1) { while (1) {
pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter); pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
...@@ -885,6 +891,33 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) { ...@@ -885,6 +891,33 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
qDebug("delete req enqueue stream task: %d, ver: %" PRId64, pTask->taskId, ver); qDebug("delete req enqueue stream task: %d, ver: %" PRId64, pTask->taskId, ver);
if (!failed) {
SStreamRefDataBlock* pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM);
pRefBlock->type = STREAM_INPUT__REF_DATA_BLOCK;
pRefBlock->pBlock = pDelBlock;
pRefBlock->dataRef = pRef;
atomic_add_fetch_32(pRefBlock->dataRef, 1);
if (streamTaskInput(pTask, (SStreamQueueItem*)pRefBlock) < 0) {
qError("stream task input del failed, task id %d", pTask->taskId);
continue;
}
if (streamSchedExec(pTask) < 0) {
qError("stream task launch failed, task id %d", pTask->taskId);
continue;
}
} else {
streamTaskInputFail(pTask);
}
}
int32_t ref = atomic_sub_fetch_32(pRef, 1);
ASSERT(ref >= 0);
if (ref == 0) {
taosMemoryFree(pDelBlock);
taosMemoryFree(pRef);
}
#if 0
SStreamDataBlock* pStreamBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); SStreamDataBlock* pStreamBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
pStreamBlock->type = STREAM_INPUT__DATA_BLOCK; pStreamBlock->type = STREAM_INPUT__DATA_BLOCK;
pStreamBlock->blocks = taosArrayInit(0, sizeof(SSDataBlock)); pStreamBlock->blocks = taosArrayInit(0, sizeof(SSDataBlock));
...@@ -908,6 +941,7 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) { ...@@ -908,6 +941,7 @@ int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
} }
} }
blockDataDestroy(pDelBlock); blockDataDestroy(pDelBlock);
#endif
return 0; return 0;
} }
...@@ -1045,6 +1079,7 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) { ...@@ -1045,6 +1079,7 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {
SDecoder decoder; SDecoder decoder;
tDecoderInit(&decoder, msgBody, msgLen); tDecoderInit(&decoder, msgBody, msgLen);
tDecodeStreamRetrieveReq(&decoder, &req); tDecodeStreamRetrieveReq(&decoder, &req);
tDecoderClear(&decoder);
int32_t taskId = req.dstTaskId; int32_t taskId = req.dstTaskId;
SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId); SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId);
if (pTask) { if (pTask) {
...@@ -1053,6 +1088,7 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) { ...@@ -1053,6 +1088,7 @@ int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {
.code = 0, .code = 0,
}; };
streamProcessRetrieveReq(pTask, &req, &rsp); streamProcessRetrieveReq(pTask, &req, &rsp);
tDeleteStreamRetrieveReq(&req);
return 0; return 0;
} else { } else {
return -1; return -1;
......
source/libs/executor/src/scanoperator.c
浏览文件 @ b9a248d7
...@@ -1480,6 +1480,40 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { ...@@ -1480,6 +1480,40 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
} }
} }
static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, SStreamScanInfo* pInfo) {
STqReader* pReader = pInfo->tqReader;
int32_t rows = pSrc->info.rows;
blockDataEnsureCapacity(pDst, rows);
SColumnInfoData* pSrcStartCol = taosArrayGet(pSrc->pDataBlock, START_TS_COLUMN_INDEX);
uint64_t* startCol = (uint64_t*)pSrcStartCol->pData;
SColumnInfoData* pSrcEndCol = taosArrayGet(pSrc->pDataBlock, END_TS_COLUMN_INDEX);
uint64_t* endCol = (uint64_t*)pSrcEndCol->pData;
SColumnInfoData* pSrcUidCol = taosArrayGet(pSrc->pDataBlock, UID_COLUMN_INDEX);
uint64_t* uidCol = (uint64_t*)pSrcUidCol->pData;
SColumnInfoData* pDstStartCol = taosArrayGet(pDst->pDataBlock, START_TS_COLUMN_INDEX);
SColumnInfoData* pDstEndCol = taosArrayGet(pDst->pDataBlock, END_TS_COLUMN_INDEX);
SColumnInfoData* pDstUidCol = taosArrayGet(pDst->pDataBlock, UID_COLUMN_INDEX);
int32_t j = 0;
for (int32_t i = 0; i < rows; i++) {
if (taosHashGet(pReader->tbIdHash, &uidCol[i], sizeof(uint64_t))) {
colDataAppend(pDstStartCol, j, (const char*)&startCol[i], false);
colDataAppend(pDstEndCol, j, (const char*)&endCol[i], false);
colDataAppend(pDstUidCol, j, (const char*)&uidCol[i], false);
colDataAppendNULL(taosArrayGet(pDst->pDataBlock, GROUPID_COLUMN_INDEX), j);
colDataAppendNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), j);
colDataAppendNULL(taosArrayGet(pDst->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), j);
j++;
}
}
pDst->info = pSrc->info;
pDst->info.rows = j;
return 0;
}
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
// NOTE: this operator does never check if current status is done or not // NOTE: this operator does never check if current status is done or not
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
...@@ -1568,6 +1602,12 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { ...@@ -1568,6 +1602,12 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
} break; } break;
case STREAM_DELETE_DATA: { case STREAM_DELETE_DATA: {
printDataBlock(pBlock, "stream scan delete recv"); printDataBlock(pBlock, "stream scan delete recv");
if (pInfo->tqReader) {
SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
filterDelBlockByUid(pDelBlock, pBlock, pInfo);
pBlock = pDelBlock;
}
printDataBlock(pBlock, "stream scan delete recv filtered");
if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) { if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) {
generateDeleteResultBlock(pInfo, pBlock, pInfo->pDeleteDataRes); generateDeleteResultBlock(pInfo, pBlock, pInfo->pDeleteDataRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT; pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;
......
source/libs/stream/src/stream.c
浏览文件 @ b9a248d7
...@@ -182,7 +182,7 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SStreamDispatchReq* pReq, S ...@@ -182,7 +182,7 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SStreamDispatchReq* pReq, S
pReq->upstreamTaskId); pReq->upstreamTaskId);
streamTaskEnqueue(pTask, pReq, pRsp); streamTaskEnqueue(pTask, pReq, pRsp);
tFreeStreamDispatchReq(pReq); tDeleteStreamDispatchReq(pReq);
if (exec) { if (exec) {
if (streamTryExec(pTask) < 0) { if (streamTryExec(pTask) < 0) {
......
source/libs/stream/src/streamData.c
浏览文件 @ b9a248d7
...@@ -179,5 +179,15 @@ void streamFreeQitem(SStreamQueueItem* data) { ...@@ -179,5 +179,15 @@ void streamFreeQitem(SStreamQueueItem* data) {
taosArrayDestroy(pMerge->reqs); taosArrayDestroy(pMerge->reqs);
taosArrayDestroy(pMerge->dataRefs); taosArrayDestroy(pMerge->dataRefs);
taosFreeQitem(pMerge); taosFreeQitem(pMerge);
} else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;
int32_t ref = atomic_sub_fetch_32(pRefBlock->dataRef, 1);
ASSERT(ref >= 0);
if (ref == 0) {
blockDataDestroy(pRefBlock->pBlock);
taosMemoryFree(pRefBlock->dataRef);
}
taosFreeQitem(pRefBlock);
} }
} }
source/libs/stream/src/streamDispatch.c
浏览文件 @ b9a248d7
...@@ -62,7 +62,7 @@ int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) { ...@@ -62,7 +62,7 @@ int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq) {
return 0; return 0;
} }
void tFreeStreamDispatchReq(SStreamDispatchReq* pReq) { void tDeleteStreamDispatchReq(SStreamDispatchReq* pReq) {
taosArrayDestroyP(pReq->data, taosMemoryFree); taosArrayDestroyP(pReq->data, taosMemoryFree);
taosArrayDestroy(pReq->dataLen); taosArrayDestroy(pReq->dataLen);
} }
...@@ -95,7 +95,10 @@ int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) { ...@@ -95,7 +95,10 @@ int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq) {
return 0; return 0;
} }
void tDeleteStreamRetrieveReq(SStreamRetrieveReq* pReq) { taosMemoryFree(pReq->pRetrieve); }
int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) { int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) {
int32_t code = -1;
SRetrieveTableRsp* pRetrieve = NULL; SRetrieveTableRsp* pRetrieve = NULL;
void* buf = NULL; void* buf = NULL;
int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock); int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
...@@ -143,7 +146,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) ...@@ -143,7 +146,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
buf = rpcMallocCont(sizeof(SMsgHead) + len); buf = rpcMallocCont(sizeof(SMsgHead) + len);
if (buf == NULL) { if (buf == NULL) {
goto FAIL; goto CLEAR;
} }
((SMsgHead*)buf)->vgId = htonl(pEpInfo->nodeId); ((SMsgHead*)buf)->vgId = htonl(pEpInfo->nodeId);
...@@ -151,6 +154,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) ...@@ -151,6 +154,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
SEncoder encoder; SEncoder encoder;
tEncoderInit(&encoder, abuf, len); tEncoderInit(&encoder, abuf, len);
tEncodeStreamRetrieveReq(&encoder, &req); tEncodeStreamRetrieveReq(&encoder, &req);
tEncoderClear(&encoder);
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
.code = 0, .code = 0,
...@@ -161,17 +165,18 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) ...@@ -161,17 +165,18 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
if (tmsgSendReq(&pEpInfo->epSet, &rpcMsg) < 0) { if (tmsgSendReq(&pEpInfo->epSet, &rpcMsg) < 0) {
ASSERT(0); ASSERT(0);
return -1; goto CLEAR;
} }
buf = NULL;
qDebug("task %d(child %d) send retrieve req to task %d at node %d, reqId %" PRId64, pTask->taskId, qDebug("task %d(child %d) send retrieve req to task %d at node %d, reqId %" PRId64, pTask->taskId,
pTask->selfChildId, pEpInfo->taskId, pEpInfo->nodeId, req.reqId); pTask->selfChildId, pEpInfo->taskId, pEpInfo->nodeId, req.reqId);
} }
return 0; code = 0;
FAIL: CLEAR:
if (pRetrieve) taosMemoryFree(pRetrieve); taosMemoryFree(pRetrieve);
if (buf) taosMemoryFree(buf); rpcFreeCont(buf);
return -1; return code;
} }
static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDispatchReq* pReq) { static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDispatchReq* pReq) {
......
source/libs/stream/src/streamExec.c
浏览文件 @ b9a248d7
...@@ -38,6 +38,9 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray* ...@@ -38,6 +38,9 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, const void* data, SArray*
SArray* blocks = pMerged->reqs; SArray* blocks = pMerged->reqs;
qDebug("task %d %p set submit input (merged), batch num: %d", pTask->taskId, pTask, (int32_t)blocks->size); qDebug("task %d %p set submit input (merged), batch num: %d", pTask->taskId, pTask, (int32_t)blocks->size);
qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_INPUT__MERGED_SUBMIT); qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_INPUT__MERGED_SUBMIT);
} else if (pItem->type == STREAM_INPUT__REF_DATA_BLOCK) {
const SStreamRefDataBlock* pRefBlock = (const SStreamRefDataBlock*)data;
qSetMultiStreamInput(exec, pRefBlock->pBlock, 1, STREAM_INPUT__DATA_BLOCK);
} else { } else {
ASSERT(0); ASSERT(0);
} }
......
tools/shell/src/shellUtil.c
浏览文件 @ b9a248d7
...@@ -143,7 +143,7 @@ void shellCheckConnectMode() { ...@@ -143,7 +143,7 @@ void shellCheckConnectMode() {
shell.args.port = 6041; shell.args.port = 6041;
} }
shell.args.dsn = taosMemoryCalloc(1, 1024); shell.args.dsn = taosMemoryCalloc(1, 1024);
snprintf(shell.args.dsn, 1024, "ws://%s:%d/rest/ws", snprintf(shell.args.dsn, 1024, "ws://%s:%d",
shell.args.host, shell.args.port); shell.args.host, shell.args.port);
} }
shell.args.cloud = false; shell.args.cloud = false;
......
tools/shell/src/shellWebsocket.c
浏览文件 @ b9a248d7
...@@ -206,26 +206,31 @@ void shellRunSingleCommandWebsocketImp(char *command) { ...@@ -206,26 +206,31 @@ void shellRunSingleCommandWebsocketImp(char *command) {
printMode = true; // When output to a file, the switch does not work. printMode = true; // When output to a file, the switch does not work.
} }
if (!shell.ws_conn && shell_conn_ws_server(0)) {
return;
}
shell.stop_query = false; shell.stop_query = false;
st = taosGetTimestampUs(); WS_RES* res;
WS_RES* res = ws_query_timeout(shell.ws_conn, command, shell.args.timeout); for (int reconnectNum = 0; reconnectNum < 2; reconnectNum++) {
int code = ws_errno(res); if (!shell.ws_conn && shell_conn_ws_server(0)) {
if (code != 0) { return;
et = taosGetTimestampUs();
fprintf(stderr, "\nDB: error: %s (%.6fs)\n", ws_errstr(res), (et - st)/1E6);
if (code == TSDB_CODE_WS_SEND_TIMEOUT || code == TSDB_CODE_WS_RECV_TIMEOUT) {
fprintf(stderr, "Hint: use -t to increase the timeout in seconds\n");
} else if (code == TSDB_CODE_WS_INTERNAL_ERRO || code == TSDB_CODE_WS_CLOSED) {
fprintf(stderr, "TDengine server is down, will try to reconnect\n");
shell.ws_conn = NULL;
} }
ws_free_result(res); st = taosGetTimestampUs();
return;
res = ws_query_timeout(shell.ws_conn, command, shell.args.timeout);
int code = ws_errno(res);
if (code != 0 && !shell.stop_query) {
et = taosGetTimestampUs();
fprintf(stderr, "\nDB: error: %s (%.6fs)\n", ws_errstr(res), (et - st)/1E6);
if (code == TSDB_CODE_WS_SEND_TIMEOUT || code == TSDB_CODE_WS_RECV_TIMEOUT) {
fprintf(stderr, "Hint: use -t to increase the timeout in seconds\n");
} else if (code == TSDB_CODE_WS_INTERNAL_ERRO || code == TSDB_CODE_WS_CLOSED) {
fprintf(stderr, "TDengine server is down, will try to reconnect\n");
shell.ws_conn = NULL;
}
ws_free_result(res);
if (reconnectNum == 0) continue;
return;
}
break;
} }
double execute_time = ws_take_timing(res)/1E6; double execute_time = ws_take_timing(res)/1E6;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或