fix:[TD-28567] do not add md5 to the end of subtable if create stream by sma
This commit is contained in:
wangmm0220
parent
8a4b79bf2c
commit
ac9582b24a
|
|
@ -91,6 +91,7 @@ TDengine 会为 WAL 文件自动创建索引以支持快速随机访问，并提
|
|||
不同语言下， TMQ 订阅相关的 API 及数据结构如下（详细的接口说明可以参考连接器章节，注意consumer结构不是线程安全的，在一个线程使用consumer时，不要在另一个线程close这个consumer）：
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<TabItem value="c" label="C">
|
||||
|
||||
```c
|
||||
|
|
@ -146,7 +147,6 @@ TDengine 会为 WAL 文件自动创建索引以支持快速随机访问，并提
|
|||
DLL_EXPORT int64_t tmq_get_vgroup_offset(TAOS_RES* res);
|
||||
DLL_EXPORT const char *tmq_err2str(int32_t code);
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
|
||||
|
|
@ -304,39 +304,38 @@ TDengine 会为 WAL 文件自动创建索引以支持快速随机访问，并提
|
|||
|
||||
void Close()
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</Tabs>
|
||||
</Tabs>
|
||||
|
||||
# 数据订阅示例
|
||||
## 写入数据
|
||||
# 数据订阅示例
|
||||
## 写入数据
|
||||
|
||||
首先完成建库、建一张超级表和多张子表操作，然后就可以写入数据了，比如：
|
||||
首先完成建库、建一张超级表和多张子表操作，然后就可以写入数据了，比如：
|
||||
|
||||
```sql
|
||||
DROP DATABASE IF EXISTS tmqdb;
|
||||
CREATE DATABASE tmqdb WAL_RETENTION_PERIOD 3600;
|
||||
CREATE TABLE tmqdb.stb (ts TIMESTAMP, c1 INT, c2 FLOAT, c3 VARCHAR(16)) TAGS(t1 INT, t3 VARCHAR(16));
|
||||
CREATE TABLE tmqdb.ctb0 USING tmqdb.stb TAGS(0, "subtable0");
|
||||
CREATE TABLE tmqdb.ctb1 USING tmqdb.stb TAGS(1, "subtable1");
|
||||
INSERT INTO tmqdb.ctb0 VALUES(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00');
|
||||
```sql
|
||||
DROP DATABASE IF EXISTS tmqdb;
|
||||
CREATE DATABASE tmqdb WAL_RETENTION_PERIOD 3600;
|
||||
CREATE TABLE tmqdb.stb (ts TIMESTAMP, c1 INT, c2 FLOAT, c3 VARCHAR(16)) TAGS(t1 INT, t3 VARCHAR(16));
|
||||
CREATE TABLE tmqdb.ctb0 USING tmqdb.stb TAGS(0, "subtable0");
|
||||
CREATE TABLE tmqdb.ctb1 USING tmqdb.stb TAGS(1, "subtable1");
|
||||
INSERT INTO tmqdb.ctb0 VALUES(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00');
|
||||
INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
||||
```
|
||||
## 创建 topic
|
||||
|
||||
使用 SQL 创建一个 topic：
|
||||
使用 SQL 创建一个 topic：
|
||||
|
||||
```sql
|
||||
CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHERE c1 > 1;
|
||||
```
|
||||
```sql
|
||||
CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHERE c1 > 1;
|
||||
```
|
||||
|
||||
## 创建消费者 *consumer*
|
||||
## 创建消费者 *consumer*
|
||||
|
||||
对于不同编程语言，其设置方式如下：
|
||||
对于不同编程语言，其设置方式如下：
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
<TabItem value="c" label="C">
|
||||
|
||||
```c
|
||||
/* 根据需要，设置消费组 (group.id)、自动提交 (enable.auto.commit)、
|
||||
自动提交时间间隔 (auto.commit.interval.ms)、用户名 (td.connect.user)、密码 (td.connect.pass) 等参数 */
|
||||
|
|
@ -353,8 +352,8 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
|
||||
tmq_conf_destroy(conf);
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
|
||||
<TabItem value="java" label="Java">
|
||||
|
||||
对于 Java 程序，还可以使用如下配置项：
|
||||
|
|
@ -388,7 +387,6 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
public class MetersDeserializer extends ReferenceDeserializer<Meters> {
|
||||
}
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Go" value="Go">
|
||||
|
|
@ -459,7 +457,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
// 自动提交时间间隔 (auto.commit.interval.ms)、用户名 (td.connect.user)、密码 (td.connect.pass) 等参数
|
||||
|
||||
let consumer = taos.consumer({
|
||||
'enable.auto.commit': 'true',
|
||||
'enable.auto.commit': 'true',
|
||||
'auto.commit.interval.ms','1000',
|
||||
'group.id': 'tg2',
|
||||
'td.connect.user': 'root',
|
||||
|
|
@ -471,9 +469,9 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
});
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem value="C#" label="C#">
|
||||
<TabItem value="C#" label="C#">
|
||||
|
||||
```csharp
|
||||
var cfg = new Dictionary<string, string>()
|
||||
|
|
@ -493,18 +491,19 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
</TabItem>
|
||||
|
||||
</Tabs>
|
||||
</Tabs>
|
||||
|
||||
上述配置中包括 consumer group ID，如果多个 consumer 指定的 consumer group ID 一样，则自动形成一个 consumer group，共享消费进度。
|
||||
上述配置中包括 consumer group ID，如果多个 consumer 指定的 consumer group ID 一样，则自动形成一个 consumer group，共享消费进度。
|
||||
|
||||
## 订阅 *topics*
|
||||
## 订阅 *topics*
|
||||
|
||||
一个 consumer 支持同时订阅多个 topic。
|
||||
一个 consumer 支持同时订阅多个 topic。
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
<TabItem value="c" label="C">
|
||||
|
||||
```c
|
||||
```c
|
||||
// 创建订阅 topics 列表
|
||||
tmq_list_t* topicList = tmq_list_new();
|
||||
tmq_list_append(topicList, "topicName");
|
||||
|
|
@ -514,8 +513,8 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
|
||||
```java
|
||||
List<String> topics = new ArrayList<>();
|
||||
|
|
@ -523,8 +522,8 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
consumer.subscribe(topics);
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
<TabItem value="Go" label="Go">
|
||||
</TabItem>
|
||||
<TabItem value="Go" label="Go">
|
||||
|
||||
```go
|
||||
err = consumer.Subscribe("example_tmq_topic", nil)
|
||||
|
|
@ -536,7 +535,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
</TabItem>
|
||||
<TabItem value="Rust" label="Rust">
|
||||
|
||||
```rust
|
||||
```rust
|
||||
consumer.subscribe(["tmq_meters"]).await?;
|
||||
```
|
||||
|
||||
|
|
@ -544,7 +543,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="Python" label="Python">
|
||||
|
||||
```python
|
||||
```python
|
||||
consumer.subscribe(['topic1', 'topic2'])
|
||||
```
|
||||
|
||||
|
|
@ -552,7 +551,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
|
||||
```js
|
||||
```js
|
||||
// 创建订阅 topics 列表
|
||||
let topics = ['topic_test']
|
||||
|
||||
|
|
@ -564,7 +563,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="C#" label="C#">
|
||||
|
||||
```csharp
|
||||
```csharp
|
||||
// 创建订阅 topics 列表
|
||||
List<String> topics = new List<string>();
|
||||
topics.add("tmq_topic");
|
||||
|
|
@ -574,16 +573,17 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
</TabItem>
|
||||
|
||||
</Tabs>
|
||||
</Tabs>
|
||||
|
||||
## 消费
|
||||
## 消费
|
||||
|
||||
以下代码展示了不同语言下如何对 TMQ 消息进行消费。
|
||||
以下代码展示了不同语言下如何对 TMQ 消息进行消费。
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
<TabItem value="c" label="C">
|
||||
|
||||
```c
|
||||
```c
|
||||
// 消费数据
|
||||
while (running) {
|
||||
TAOS_RES* msg = tmq_consumer_poll(tmq, timeOut);
|
||||
|
|
@ -591,10 +591,10 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
}
|
||||
```
|
||||
|
||||
这里是一个 **while** 循环，每调用一次 tmq_consumer_poll()，获取一个消息，该消息与普通查询返回的结果集完全相同，可以使用相同的解析 API 完成消息内容的解析。
|
||||
这里是一个 **while** 循环，每调用一次 tmq_consumer_poll()，获取一个消息，该消息与普通查询返回的结果集完全相同，可以使用相同的解析 API 完成消息内容的解析。
|
||||
|
||||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
|
||||
```java
|
||||
while(running){
|
||||
|
|
@ -605,9 +605,9 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
}
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem value="Go" label="Go">
|
||||
<TabItem value="Go" label="Go">
|
||||
|
||||
```go
|
||||
for {
|
||||
|
|
@ -625,9 +625,9 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
}
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem value="Rust" label="Rust">
|
||||
<TabItem value="Rust" label="Rust">
|
||||
|
||||
```rust
|
||||
{
|
||||
|
|
@ -660,8 +660,8 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
}
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
<TabItem value="Python" label="Python">
|
||||
</TabItem>
|
||||
<TabItem value="Python" label="Python">
|
||||
|
||||
```python
|
||||
while True:
|
||||
|
|
@ -677,9 +677,9 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
print(block.fetchall())
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
|
||||
```js
|
||||
while(true){
|
||||
|
|
@ -695,7 +695,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="C#" label="C#">
|
||||
|
||||
```csharp
|
||||
```csharp
|
||||
// 消费数据
|
||||
while (true)
|
||||
{
|
||||
|
|
@ -710,16 +710,17 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
</TabItem>
|
||||
|
||||
</Tabs>
|
||||
</Tabs>
|
||||
|
||||
## 结束消费
|
||||
## 结束消费
|
||||
|
||||
消费结束后，应当取消订阅。
|
||||
消费结束后，应当取消订阅。
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
<TabItem value="c" label="C">
|
||||
|
||||
```c
|
||||
```c
|
||||
/* 取消订阅 */
|
||||
tmq_unsubscribe(tmq);
|
||||
|
||||
|
|
@ -730,7 +731,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
</TabItem>
|
||||
<TabItem value="java" label="Java">
|
||||
|
||||
```java
|
||||
```java
|
||||
/* 取消订阅 */
|
||||
consumer.unsubscribe();
|
||||
|
||||
|
|
@ -742,7 +743,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="Go" label="Go">
|
||||
|
||||
```go
|
||||
```go
|
||||
/* Unsubscribe */
|
||||
_ = consumer.Unsubscribe()
|
||||
|
||||
|
|
@ -754,7 +755,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="Rust" label="Rust">
|
||||
|
||||
```rust
|
||||
```rust
|
||||
consumer.unsubscribe().await;
|
||||
```
|
||||
|
||||
|
|
@ -762,7 +763,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="Python" label="Python">
|
||||
|
||||
```py
|
||||
```py
|
||||
# 取消订阅
|
||||
consumer.unsubscribe()
|
||||
# 关闭消费
|
||||
|
|
@ -772,7 +773,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
</TabItem>
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
|
||||
```js
|
||||
```js
|
||||
consumer.unsubscribe();
|
||||
consumer.close();
|
||||
```
|
||||
|
|
@ -781,7 +782,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<TabItem value="C#" label="C#">
|
||||
|
||||
```csharp
|
||||
```csharp
|
||||
// 取消订阅
|
||||
consumer.Unsubscribe();
|
||||
|
||||
|
|
@ -789,7 +790,7 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
consumer.Close();
|
||||
```
|
||||
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
</Tabs>
|
||||
|
||||
|
|
@ -799,40 +800,40 @@ INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');
|
|||
|
||||
<Tabs defaultValue="java" groupId="lang">
|
||||
|
||||
<TabItem label="C" value="c">
|
||||
<TabItem label="C" value="c">
|
||||
<CDemo />
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Java" value="java">
|
||||
<Tabs defaultValue="native">
|
||||
<TabItem value="native" label="本地连接">
|
||||
<Java />
|
||||
</TabItem>
|
||||
<TabItem value="ws" label="WebSocket 连接">
|
||||
<JavaWS />
|
||||
</TabItem>
|
||||
</Tabs>
|
||||
</TabItem>
|
||||
<TabItem label="Java" value="java">
|
||||
<Tabs defaultValue="native">
|
||||
<TabItem value="native" label="本地连接">
|
||||
<Java />
|
||||
</TabItem>
|
||||
<TabItem value="ws" label="WebSocket 连接">
|
||||
<JavaWS />
|
||||
</TabItem>
|
||||
</Tabs>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Go" value="Go">
|
||||
<TabItem label="Go" value="Go">
|
||||
<Go/>
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Rust" value="Rust">
|
||||
<TabItem label="Rust" value="Rust">
|
||||
<Rust />
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Python" value="Python">
|
||||
<TabItem label="Python" value="Python">
|
||||
<Python />
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
<TabItem label="Node.JS" value="Node.JS">
|
||||
<Node/>
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
<TabItem label="C#" value="C#">
|
||||
<TabItem label="C#" value="C#">
|
||||
<CSharp/>
|
||||
</TabItem>
|
||||
</TabItem>
|
||||
|
||||
</Tabs>
|
||||
|
||||
|
|
|
|||
|
|
@ -1414,7 +1414,7 @@ typedef struct {
|
|||
|
||||
int32_t tSerializeSKillCompactReq(void* buf, int32_t bufLen, SKillCompactReq* pReq);
|
||||
int32_t tDeserializeSKillCompactReq(void* buf, int32_t bufLen, SKillCompactReq* pReq);
|
||||
void tFreeSKillCompactReq(SKillCompactReq *pReq);
|
||||
void tFreeSKillCompactReq(SKillCompactReq* pReq);
|
||||
|
||||
typedef struct {
|
||||
char name[TSDB_FUNC_NAME_LEN];
|
||||
|
|
|
|||
|
|
@ -463,7 +463,8 @@ struct SStreamTask {
|
|||
struct SStreamMeta* pMeta;
|
||||
SSHashObj* pNameMap;
|
||||
void* pBackend;
|
||||
char reserve[256];
|
||||
int8_t subtableWithoutMd5;
|
||||
char reserve[255];
|
||||
};
|
||||
|
||||
typedef int32_t (*startComplete_fn_t)(struct SStreamMeta*);
|
||||
|
|
@ -532,7 +533,7 @@ int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo)
|
|||
int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo);
|
||||
|
||||
SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, SEpSet* pEpset, bool fillHistory, int64_t triggerParam,
|
||||
SArray* pTaskList, bool hasFillhistory);
|
||||
SArray* pTaskList, bool hasFillhistory, int8_t subtableWithoutMd5);
|
||||
int32_t tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
|
||||
int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
|
||||
void tFreeStreamTask(SStreamTask* pTask);
|
||||
|
|
@ -839,7 +840,8 @@ SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st);
|
|||
// stream task meta
|
||||
void streamMetaInit();
|
||||
void streamMetaCleanup();
|
||||
SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId, int64_t stage, startComplete_fn_t fn);
|
||||
SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId, int64_t stage,
|
||||
startComplete_fn_t fn);
|
||||
void streamMetaClose(SStreamMeta* streamMeta);
|
||||
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask); // save to stream meta store
|
||||
int32_t streamMetaRemoveTask(SStreamMeta* pMeta, STaskId* pKey);
|
||||
|
|
@ -890,8 +892,8 @@ int32_t buildCheckpointSourceRsp(SStreamCheckpointSourceReq* pReq, SRpcHandleInf
|
|||
SStreamTaskSM* streamCreateStateMachine(SStreamTask* pTask);
|
||||
void* streamDestroyStateMachine(SStreamTaskSM* pSM);
|
||||
|
||||
int32_t broadcastRetrieveMsg(SStreamTask* pTask, SStreamRetrieveReq *req);
|
||||
void sendRetrieveRsp(SStreamRetrieveReq *pReq, SRpcMsg* pRsp);
|
||||
int32_t broadcastRetrieveMsg(SStreamTask* pTask, SStreamRetrieveReq* req);
|
||||
void sendRetrieveRsp(SStreamRetrieveReq* pReq, SRpcMsg* pRsp);
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
|
|
@ -699,6 +699,7 @@ typedef struct {
|
|||
int64_t checkpointId;
|
||||
|
||||
int32_t indexForMultiAggBalance;
|
||||
int8_t subTableWithoutMd5;
|
||||
char reserve[256];
|
||||
|
||||
} SStreamObj;
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ extern "C" {
|
|||
#endif
|
||||
|
||||
#define MND_STREAM_RESERVE_SIZE 64
|
||||
#define MND_STREAM_VER_NUMBER 4
|
||||
#define MND_STREAM_VER_NUMBER 5
|
||||
|
||||
#define MND_STREAM_CREATE_NAME "stream-create"
|
||||
#define MND_STREAM_CHECKPOINT_NAME "stream-checkpoint"
|
||||
|
|
|
|||
|
|
@ -85,6 +85,7 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
|
|||
|
||||
// 3.0.50 ver = 3
|
||||
if (tEncodeI64(pEncoder, pObj->checkpointId) < 0) return -1;
|
||||
if (tEncodeI8(pEncoder, pObj->subTableWithoutMd5) < 0) return -1;
|
||||
|
||||
if (tEncodeCStrWithLen(pEncoder, pObj->reserve, sizeof(pObj->reserve) - 1) < 0) return -1;
|
||||
|
||||
|
|
@ -168,6 +169,10 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {
|
|||
if (sver >= 3) {
|
||||
if (tDecodeI64(pDecoder, &pObj->checkpointId) < 0) return -1;
|
||||
}
|
||||
|
||||
if (sver >= 5) {
|
||||
if (tDecodeI8(pDecoder, &pObj->subTableWithoutMd5) < 0) return -1;
|
||||
}
|
||||
if (tDecodeCStrTo(pDecoder, pObj->reserve) < 0) return -1;
|
||||
|
||||
tEndDecode(pDecoder);
|
||||
|
|
|
|||
|
|
@ -14,13 +14,13 @@
|
|||
*/
|
||||
|
||||
#include "mndScheduler.h"
|
||||
#include "tmisce.h"
|
||||
#include "mndMnode.h"
|
||||
#include "mndDb.h"
|
||||
#include "mndMnode.h"
|
||||
#include "mndSnode.h"
|
||||
#include "mndVgroup.h"
|
||||
#include "parser.h"
|
||||
#include "tcompare.h"
|
||||
#include "tmisce.h"
|
||||
#include "tname.h"
|
||||
#include "tuuid.h"
|
||||
|
||||
|
|
@ -189,7 +189,7 @@ SVgObj* mndSchedFetchOneVg(SMnode* pMnode, SStreamObj* pStream) {
|
|||
return NULL;
|
||||
}
|
||||
|
||||
if(pStream->indexForMultiAggBalance == -1){
|
||||
if (pStream->indexForMultiAggBalance == -1) {
|
||||
taosSeedRand(taosSafeRand());
|
||||
pStream->indexForMultiAggBalance = taosRand() % pDbObj->cfg.numOfVgroups;
|
||||
}
|
||||
|
|
@ -204,7 +204,7 @@ SVgObj* mndSchedFetchOneVg(SMnode* pMnode, SStreamObj* pStream) {
|
|||
sdbRelease(pMnode->pSdb, pVgroup);
|
||||
continue;
|
||||
}
|
||||
if (index++ == pStream->indexForMultiAggBalance){
|
||||
if (index++ == pStream->indexForMultiAggBalance) {
|
||||
pStream->indexForMultiAggBalance++;
|
||||
pStream->indexForMultiAggBalance %= pDbObj->cfg.numOfVgroups;
|
||||
sdbCancelFetch(pMnode->pSdb, pIter);
|
||||
|
|
@ -217,12 +217,12 @@ SVgObj* mndSchedFetchOneVg(SMnode* pMnode, SStreamObj* pStream) {
|
|||
return pVgroup;
|
||||
}
|
||||
|
||||
static int32_t doAddSinkTask(SStreamObj* pStream, SMnode* pMnode, SVgObj* pVgroup,
|
||||
SEpSet* pEpset, bool isFillhistory) {
|
||||
static int32_t doAddSinkTask(SStreamObj* pStream, SMnode* pMnode, SVgObj* pVgroup, SEpSet* pEpset, bool isFillhistory) {
|
||||
int64_t uid = (isFillhistory) ? pStream->hTaskUid : pStream->uid;
|
||||
SArray** pTaskList = (isFillhistory) ? taosArrayGetLast(pStream->pHTasksList) : taosArrayGetLast(pStream->tasks);
|
||||
|
||||
SStreamTask* pTask = tNewStreamTask(uid, TASK_LEVEL__SINK, pEpset, isFillhistory, 0, *pTaskList, pStream->conf.fillHistory);
|
||||
SStreamTask* pTask = tNewStreamTask(uid, TASK_LEVEL__SINK, pEpset, isFillhistory, 0, *pTaskList,
|
||||
pStream->conf.fillHistory, pStream->subTableWithoutMd5);
|
||||
if (pTask == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return terrno;
|
||||
|
|
@ -235,12 +235,12 @@ static int32_t doAddSinkTask(SStreamObj* pStream, SMnode* pMnode, SVgObj* pVgrou
|
|||
return mndSetSinkTaskInfo(pStream, pTask);
|
||||
}
|
||||
|
||||
static int32_t doAddSinkTaskToVg(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset, SVgObj* vgObj){
|
||||
static int32_t doAddSinkTaskToVg(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset, SVgObj* vgObj) {
|
||||
int32_t code = doAddSinkTask(pStream, pMnode, vgObj, pEpset, false);
|
||||
if (code != 0) {
|
||||
return code;
|
||||
}
|
||||
if(pStream->conf.fillHistory){
|
||||
if (pStream->conf.fillHistory) {
|
||||
code = doAddSinkTask(pStream, pMnode, vgObj, pEpset, true);
|
||||
if (code != 0) {
|
||||
return code;
|
||||
|
|
@ -267,7 +267,7 @@ static int32_t doAddShuffleSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet*
|
|||
}
|
||||
|
||||
int32_t code = doAddSinkTaskToVg(pMnode, pStream, pEpset, pVgroup);
|
||||
if(code != 0){
|
||||
if (code != 0) {
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
return code;
|
||||
}
|
||||
|
|
@ -279,7 +279,7 @@ static int32_t doAddShuffleSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet*
|
|||
}
|
||||
|
||||
static int64_t getVgroupLastVer(const SArray* pList, int32_t vgId) {
|
||||
for(int32_t i = 0; i < taosArrayGetSize(pList); ++i) {
|
||||
for (int32_t i = 0; i < taosArrayGetSize(pList); ++i) {
|
||||
SVgroupVer* pVer = taosArrayGet(pList, i);
|
||||
if (pVer->vgId == vgId) {
|
||||
return pVer->ver;
|
||||
|
|
@ -315,19 +315,18 @@ static void streamTaskSetDataRange(SStreamTask* pTask, int64_t skey, SArray* pVe
|
|||
pRange->range.minVer = latestVer + 1;
|
||||
pRange->range.maxVer = INT64_MAX;
|
||||
|
||||
mDebug("add source task 0x%x timeWindow:%" PRId64 "-%" PRId64 " verRange:%" PRId64 "-%" PRId64,
|
||||
pTask->id.taskId, pWindow->skey, pWindow->ekey, pRange->range.minVer, pRange->range.maxVer);
|
||||
mDebug("add source task 0x%x timeWindow:%" PRId64 "-%" PRId64 " verRange:%" PRId64 "-%" PRId64, pTask->id.taskId,
|
||||
pWindow->skey, pWindow->ekey, pRange->range.minVer, pRange->range.maxVer);
|
||||
}
|
||||
}
|
||||
|
||||
static SStreamTask* buildSourceTask(SStreamObj* pStream, SEpSet* pEpset,
|
||||
bool isFillhistory, bool useTriggerParam) {
|
||||
static SStreamTask* buildSourceTask(SStreamObj* pStream, SEpSet* pEpset, bool isFillhistory, bool useTriggerParam) {
|
||||
uint64_t uid = (isFillhistory) ? pStream->hTaskUid : pStream->uid;
|
||||
SArray** pTaskList = (isFillhistory) ? taosArrayGetLast(pStream->pHTasksList) : taosArrayGetLast(pStream->tasks);
|
||||
|
||||
SStreamTask* pTask = tNewStreamTask(uid, TASK_LEVEL__SOURCE, pEpset,
|
||||
isFillhistory, useTriggerParam ? pStream->conf.triggerParam : 0,
|
||||
*pTaskList, pStream->conf.fillHistory);
|
||||
SStreamTask* pTask =
|
||||
tNewStreamTask(uid, TASK_LEVEL__SOURCE, pEpset, isFillhistory, useTriggerParam ? pStream->conf.triggerParam : 0,
|
||||
*pTaskList, pStream->conf.fillHistory, pStream->subTableWithoutMd5);
|
||||
if (pTask == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
|
@ -335,7 +334,7 @@ static SStreamTask* buildSourceTask(SStreamObj* pStream, SEpSet* pEpset,
|
|||
return pTask;
|
||||
}
|
||||
|
||||
static void addNewTaskList(SStreamObj* pStream){
|
||||
static void addNewTaskList(SStreamObj* pStream) {
|
||||
SArray* pTaskList = taosArrayInit(0, POINTER_BYTES);
|
||||
taosArrayPush(pStream->tasks, &pTaskList);
|
||||
if (pStream->conf.fillHistory) {
|
||||
|
|
@ -364,11 +363,11 @@ static void setHTasksId(SStreamObj* pStream) {
|
|||
}
|
||||
}
|
||||
|
||||
static int32_t doAddSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream, SEpSet* pEpset,
|
||||
int64_t skey, SArray* pVerList, SVgObj* pVgroup, bool isFillhistory, bool useTriggerParam ){
|
||||
static int32_t doAddSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream, SEpSet* pEpset, int64_t skey,
|
||||
SArray* pVerList, SVgObj* pVgroup, bool isFillhistory, bool useTriggerParam) {
|
||||
// new stream task
|
||||
SStreamTask* pTask = buildSourceTask(pStream, pEpset, isFillhistory, useTriggerParam);
|
||||
if(pTask == NULL){
|
||||
if (pTask == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return terrno;
|
||||
}
|
||||
|
|
@ -377,14 +376,14 @@ static int32_t doAddSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStre
|
|||
streamTaskSetDataRange(pTask, skey, pVerList, pVgroup->vgId);
|
||||
|
||||
int32_t code = mndAssignStreamTaskToVgroup(pMnode, pTask, plan, pVgroup);
|
||||
if(code != 0){
|
||||
if (code != 0) {
|
||||
terrno = code;
|
||||
return terrno;
|
||||
}
|
||||
return TDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static SSubplan* getScanSubPlan(const SQueryPlan* pPlan){
|
||||
static SSubplan* getScanSubPlan(const SQueryPlan* pPlan) {
|
||||
int32_t numOfPlanLevel = LIST_LENGTH(pPlan->pSubplans);
|
||||
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, numOfPlanLevel - 1);
|
||||
if (LIST_LENGTH(inner->pNodeList) != 1) {
|
||||
|
|
@ -400,7 +399,7 @@ static SSubplan* getScanSubPlan(const SQueryPlan* pPlan){
|
|||
return plan;
|
||||
}
|
||||
|
||||
static SSubplan* getAggSubPlan(const SQueryPlan* pPlan, int index){
|
||||
static SSubplan* getAggSubPlan(const SQueryPlan* pPlan, int index) {
|
||||
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, index);
|
||||
if (LIST_LENGTH(inner->pNodeList) != 1) {
|
||||
terrno = TSDB_CODE_QRY_INVALID_INPUT;
|
||||
|
|
@ -415,8 +414,8 @@ static SSubplan* getAggSubPlan(const SQueryPlan* pPlan, int index){
|
|||
return plan;
|
||||
}
|
||||
|
||||
static int32_t addSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream,
|
||||
SEpSet* pEpset, int64_t nextWindowSkey, SArray* pVerList, bool useTriggerParam) {
|
||||
static int32_t addSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream, SEpSet* pEpset,
|
||||
int64_t nextWindowSkey, SArray* pVerList, bool useTriggerParam) {
|
||||
addNewTaskList(pStream);
|
||||
|
||||
void* pIter = NULL;
|
||||
|
|
@ -433,15 +432,16 @@ static int32_t addSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStream
|
|||
continue;
|
||||
}
|
||||
|
||||
int code = doAddSourceTask(pMnode, plan, pStream, pEpset, nextWindowSkey, pVerList, pVgroup, false, useTriggerParam);
|
||||
if(code != 0){
|
||||
int code =
|
||||
doAddSourceTask(pMnode, plan, pStream, pEpset, nextWindowSkey, pVerList, pVgroup, false, useTriggerParam);
|
||||
if (code != 0) {
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
return code;
|
||||
}
|
||||
|
||||
if (pStream->conf.fillHistory) {
|
||||
code = doAddSourceTask(pMnode, plan, pStream, pEpset, nextWindowSkey, pVerList, pVgroup, true, useTriggerParam);
|
||||
if(code != 0){
|
||||
if (code != 0) {
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
return code;
|
||||
}
|
||||
|
|
@ -461,9 +461,9 @@ static SStreamTask* buildAggTask(SStreamObj* pStream, SEpSet* pEpset, bool isFil
|
|||
uint64_t uid = (isFillhistory) ? pStream->hTaskUid : pStream->uid;
|
||||
SArray** pTaskList = (isFillhistory) ? taosArrayGetLast(pStream->pHTasksList) : taosArrayGetLast(pStream->tasks);
|
||||
|
||||
SStreamTask* pAggTask = tNewStreamTask(uid, TASK_LEVEL__AGG, pEpset, isFillhistory,
|
||||
useTriggerParam ? pStream->conf.triggerParam : 0,
|
||||
*pTaskList, pStream->conf.fillHistory);
|
||||
SStreamTask* pAggTask =
|
||||
tNewStreamTask(uid, TASK_LEVEL__AGG, pEpset, isFillhistory, useTriggerParam ? pStream->conf.triggerParam : 0,
|
||||
*pTaskList, pStream->conf.fillHistory, pStream->subTableWithoutMd5);
|
||||
if (pAggTask == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return NULL;
|
||||
|
|
@ -472,8 +472,8 @@ static SStreamTask* buildAggTask(SStreamObj* pStream, SEpSet* pEpset, bool isFil
|
|||
return pAggTask;
|
||||
}
|
||||
|
||||
static int32_t doAddAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, SEpSet* pEpset,
|
||||
SVgObj* pVgroup, SSnodeObj* pSnode, bool isFillhistory, bool useTriggerParam){
|
||||
static int32_t doAddAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, SEpSet* pEpset, SVgObj* pVgroup,
|
||||
SSnodeObj* pSnode, bool isFillhistory, bool useTriggerParam) {
|
||||
int32_t code = 0;
|
||||
SStreamTask* pTask = buildAggTask(pStream, pEpset, isFillhistory, useTriggerParam);
|
||||
if (pTask == NULL) {
|
||||
|
|
@ -490,7 +490,7 @@ static int32_t doAddAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan,
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t addAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, SEpSet* pEpset, bool useTriggerParam){
|
||||
static int32_t addAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, SEpSet* pEpset, bool useTriggerParam) {
|
||||
SVgObj* pVgroup = NULL;
|
||||
SSnodeObj* pSnode = NULL;
|
||||
int32_t code = 0;
|
||||
|
|
@ -504,20 +504,20 @@ static int32_t addAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, S
|
|||
}
|
||||
|
||||
code = doAddAggTask(pStream, pMnode, plan, pEpset, pVgroup, pSnode, false, useTriggerParam);
|
||||
if(code != 0){
|
||||
if (code != 0) {
|
||||
goto END;
|
||||
}
|
||||
|
||||
if (pStream->conf.fillHistory) {
|
||||
code = doAddAggTask(pStream, pMnode, plan, pEpset, pVgroup, pSnode, true, useTriggerParam);
|
||||
if(code != 0){
|
||||
if (code != 0) {
|
||||
goto END;
|
||||
}
|
||||
|
||||
setHTasksId(pStream);
|
||||
}
|
||||
|
||||
END:
|
||||
END:
|
||||
if (pSnode != NULL) {
|
||||
sdbRelease(pMnode->pSdb, pSnode);
|
||||
} else {
|
||||
|
|
@ -526,7 +526,7 @@ static int32_t addAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan, S
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t addSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset){
|
||||
static int32_t addSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset) {
|
||||
int32_t code = 0;
|
||||
addNewTaskList(pStream);
|
||||
|
||||
|
|
@ -548,9 +548,9 @@ static int32_t addSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset){
|
|||
return TDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static void bindTaskToSinkTask(SStreamObj* pStream, SMnode* pMnode, SArray* pSinkTaskList, SStreamTask* task){
|
||||
static void bindTaskToSinkTask(SStreamObj* pStream, SMnode* pMnode, SArray* pSinkTaskList, SStreamTask* task) {
|
||||
mndAddDispatcherForInternalTask(pMnode, pStream, pSinkTaskList, task);
|
||||
for(int32_t k = 0; k < taosArrayGetSize(pSinkTaskList); k++) {
|
||||
for (int32_t k = 0; k < taosArrayGetSize(pSinkTaskList); k++) {
|
||||
SStreamTask* pSinkTask = taosArrayGetP(pSinkTaskList, k);
|
||||
streamTaskSetUpstreamInfo(pSinkTask, task);
|
||||
}
|
||||
|
|
@ -561,7 +561,7 @@ static void bindAggSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks) {
|
|||
SArray* pSinkTaskList = taosArrayGetP(tasks, SINK_NODE_LEVEL);
|
||||
SArray** pAggTaskList = taosArrayGetLast(tasks);
|
||||
|
||||
for(int i = 0; i < taosArrayGetSize(*pAggTaskList); i++){
|
||||
for (int i = 0; i < taosArrayGetSize(*pAggTaskList); i++) {
|
||||
SStreamTask* pAggTask = taosArrayGetP(*pAggTaskList, i);
|
||||
bindTaskToSinkTask(pStream, pMnode, pSinkTaskList, pAggTask);
|
||||
mDebug("bindAggSink taskId:%s to sink task list", pAggTask->id.idStr);
|
||||
|
|
@ -572,7 +572,7 @@ static void bindSourceSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks, b
|
|||
SArray* pSinkTaskList = taosArrayGetP(tasks, SINK_NODE_LEVEL);
|
||||
SArray* pSourceTaskList = taosArrayGetP(tasks, hasExtraSink ? SINK_NODE_LEVEL + 1 : SINK_NODE_LEVEL);
|
||||
|
||||
for(int i = 0; i < taosArrayGetSize(pSourceTaskList); i++){
|
||||
for (int i = 0; i < taosArrayGetSize(pSourceTaskList); i++) {
|
||||
SStreamTask* pSourceTask = taosArrayGetP(pSourceTaskList, i);
|
||||
mDebug("bindSourceSink taskId:%s to sink task list", pSourceTask->id.idStr);
|
||||
|
||||
|
|
@ -591,8 +591,8 @@ static void bindTwoLevel(SArray* tasks, int32_t begin, int32_t end) {
|
|||
SArray* pUpTaskList = taosArrayGetP(tasks, size - 2);
|
||||
|
||||
SStreamTask** pDownTask = taosArrayGetLast(pDownTaskList);
|
||||
end = end > taosArrayGetSize(pUpTaskList) ? taosArrayGetSize(pUpTaskList): end;
|
||||
for(int i = begin; i < end; i++){
|
||||
end = end > taosArrayGetSize(pUpTaskList) ? taosArrayGetSize(pUpTaskList) : end;
|
||||
for (int i = begin; i < end; i++) {
|
||||
SStreamTask* pUpTask = taosArrayGetP(pUpTaskList, i);
|
||||
pUpTask->info.selfChildId = i - begin;
|
||||
streamTaskSetFixedDownstreamInfo(pUpTask, *pDownTask);
|
||||
|
|
@ -616,8 +616,8 @@ static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan*
|
|||
bool multiTarget = (pDbObj->cfg.numOfVgroups > 1);
|
||||
sdbRelease(pSdb, pDbObj);
|
||||
|
||||
mDebug("doScheduleStream numOfPlanLevel:%d, exDb:%d, multiTarget:%d, fix vgId:%d, physicalPlan:%s",
|
||||
numOfPlanLevel, externalTargetDB, multiTarget, pStream->fixedSinkVgId, pStream->physicalPlan);
|
||||
mDebug("doScheduleStream numOfPlanLevel:%d, exDb:%d, multiTarget:%d, fix vgId:%d, physicalPlan:%s", numOfPlanLevel,
|
||||
externalTargetDB, multiTarget, pStream->fixedSinkVgId, pStream->physicalPlan);
|
||||
pStream->tasks = taosArrayInit(numOfPlanLevel + 1, POINTER_BYTES);
|
||||
pStream->pHTasksList = taosArrayInit(numOfPlanLevel + 1, POINTER_BYTES);
|
||||
|
||||
|
|
@ -649,32 +649,32 @@ static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan*
|
|||
return TDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
if(numOfPlanLevel == 3){
|
||||
if (numOfPlanLevel == 3) {
|
||||
plan = getAggSubPlan(pPlan, 1); // middle agg plan
|
||||
if (plan == NULL) {
|
||||
return terrno;
|
||||
}
|
||||
do{
|
||||
do {
|
||||
SArray** list = taosArrayGetLast(pStream->tasks);
|
||||
float size = (float)taosArrayGetSize(*list);
|
||||
size_t cnt = (size_t)ceil(size/tsStreamAggCnt);
|
||||
if(cnt <= 1) break;
|
||||
size_t cnt = (size_t)ceil(size / tsStreamAggCnt);
|
||||
if (cnt <= 1) break;
|
||||
|
||||
mDebug("doScheduleStream add middle agg, size:%d, cnt:%d", (int)size, (int)cnt);
|
||||
addNewTaskList(pStream);
|
||||
|
||||
for(int j = 0; j < cnt; j++){
|
||||
for (int j = 0; j < cnt; j++) {
|
||||
code = addAggTask(pStream, pMnode, plan, pEpset, false);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return code;
|
||||
}
|
||||
|
||||
bindTwoLevel(pStream->tasks, j*tsStreamAggCnt, (j+1)*tsStreamAggCnt);
|
||||
bindTwoLevel(pStream->tasks, j * tsStreamAggCnt, (j + 1) * tsStreamAggCnt);
|
||||
if (pStream->conf.fillHistory) {
|
||||
bindTwoLevel(pStream->pHTasksList, j*tsStreamAggCnt, (j+1)*tsStreamAggCnt);
|
||||
bindTwoLevel(pStream->pHTasksList, j * tsStreamAggCnt, (j + 1) * tsStreamAggCnt);
|
||||
}
|
||||
}
|
||||
}while(1);
|
||||
} while (1);
|
||||
}
|
||||
|
||||
plan = getAggSubPlan(pPlan, 0);
|
||||
|
|
|
|||
|
|
@ -567,6 +567,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
|
|||
streamObj.conf.triggerParam = pCreate->maxDelay;
|
||||
streamObj.ast = taosStrdup(smaObj.ast);
|
||||
streamObj.indexForMultiAggBalance = -1;
|
||||
streamObj.subTableWithoutMd5 = 1;
|
||||
|
||||
// check the maxDelay
|
||||
if (streamObj.conf.triggerParam < TSDB_MIN_ROLLUP_MAX_DELAY) {
|
||||
|
|
|
|||
|
|
@ -192,7 +192,7 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
|
|||
STREAM_DECODE_OVER:
|
||||
taosMemoryFreeClear(buf);
|
||||
if (terrno != TSDB_CODE_SUCCESS) {
|
||||
char* p = (pStream == NULL) ? "null" : pStream->name;
|
||||
char *p = (pStream == NULL) ? "null" : pStream->name;
|
||||
mError("stream:%s, failed to decode from raw:%p since %s", p, pRaw, terrstr());
|
||||
taosMemoryFreeClear(pRow);
|
||||
return NULL;
|
||||
|
|
@ -802,8 +802,7 @@ static int32_t mndProcessStreamCheckpointTmr(SRpcMsg *pReq) {
|
|||
}
|
||||
|
||||
static int32_t mndBuildStreamCheckpointSourceReq(void **pBuf, int32_t *pLen, int32_t nodeId, int64_t checkpointId,
|
||||
int64_t streamId, int32_t taskId, int32_t transId,
|
||||
int8_t mndTrigger) {
|
||||
int64_t streamId, int32_t taskId, int32_t transId, int8_t mndTrigger) {
|
||||
SStreamCheckpointSourceReq req = {0};
|
||||
req.checkpointId = checkpointId;
|
||||
req.nodeId = nodeId;
|
||||
|
|
@ -878,11 +877,10 @@ static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStre
|
|||
int32_t code = -1;
|
||||
int64_t ts = taosGetTimestampMs();
|
||||
if (mndTrigger == 1 && (ts - pStream->checkpointFreq < tsStreamCheckpointInterval * 1000)) {
|
||||
// mWarn("checkpoint interval less than the threshold, ignore it");
|
||||
// mWarn("checkpoint interval less than the threshold, ignore it");
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
||||
bool conflict = mndStreamTransConflictCheck(pMnode, pStream->uid, MND_STREAM_CHECKPOINT_NAME, lock);
|
||||
if (conflict) {
|
||||
mndAddtoCheckpointWaitingList(pStream, checkpointId);
|
||||
|
|
@ -1144,7 +1142,7 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
|
|||
return -1;
|
||||
}
|
||||
|
||||
STrans* pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_DROP_NAME, "drop stream");
|
||||
STrans *pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_DROP_NAME, "drop stream");
|
||||
if (pTrans == NULL) {
|
||||
mError("stream:%s, failed to drop since %s", dropReq.name, terrstr());
|
||||
sdbRelease(pMnode->pSdb, pStream);
|
||||
|
|
@ -1570,7 +1568,7 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
|
|||
return -1;
|
||||
}
|
||||
|
||||
STrans* pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_PAUSE_NAME, "pause the stream");
|
||||
STrans *pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_PAUSE_NAME, "pause the stream");
|
||||
if (pTrans == NULL) {
|
||||
mError("stream:%s failed to pause stream since %s", pauseReq.name, terrstr());
|
||||
sdbRelease(pMnode->pSdb, pStream);
|
||||
|
|
@ -1590,7 +1588,7 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
|
|||
// pause stream
|
||||
taosWLockLatch(&pStream->lock);
|
||||
pStream->status = STREAM_STATUS__PAUSE;
|
||||
if (mndPersistTransLog(pStream, pTrans,SDB_STATUS_READY) < 0) {
|
||||
if (mndPersistTransLog(pStream, pTrans, SDB_STATUS_READY) < 0) {
|
||||
taosWUnLockLatch(&pStream->lock);
|
||||
|
||||
sdbRelease(pMnode->pSdb, pStream);
|
||||
|
|
@ -1617,7 +1615,7 @@ static int32_t mndProcessResumeStreamReq(SRpcMsg *pReq) {
|
|||
SMnode *pMnode = pReq->info.node;
|
||||
SStreamObj *pStream = NULL;
|
||||
|
||||
if(grantCheckExpire(TSDB_GRANT_STREAMS) < 0){
|
||||
if (grantCheckExpire(TSDB_GRANT_STREAMS) < 0) {
|
||||
terrno = TSDB_CODE_GRANT_EXPIRED;
|
||||
return -1;
|
||||
}
|
||||
|
|
@ -1659,7 +1657,7 @@ static int32_t mndProcessResumeStreamReq(SRpcMsg *pReq) {
|
|||
return -1;
|
||||
}
|
||||
|
||||
STrans* pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_RESUME_NAME, "resume the stream");
|
||||
STrans *pTrans = doCreateTrans(pMnode, pStream, pReq, MND_STREAM_RESUME_NAME, "resume the stream");
|
||||
if (pTrans == NULL) {
|
||||
mError("stream:%s, failed to resume stream since %s", resumeReq.name, terrstr());
|
||||
sdbRelease(pMnode->pSdb, pStream);
|
||||
|
|
@ -2106,10 +2104,10 @@ void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode) {
|
|||
ASSERT(taosHashGetSize(pExecNode->pTaskMap) == taosArrayGetSize(pExecNode->pTaskList));
|
||||
}
|
||||
|
||||
static void doAddTaskId(SArray* pList, int32_t taskId, int64_t uid, int32_t numOfTotal) {
|
||||
static void doAddTaskId(SArray *pList, int32_t taskId, int64_t uid, int32_t numOfTotal) {
|
||||
int32_t num = taosArrayGetSize(pList);
|
||||
for(int32_t i = 0; i < num; ++i) {
|
||||
int32_t* pId = taosArrayGet(pList, i);
|
||||
for (int32_t i = 0; i < num; ++i) {
|
||||
int32_t *pId = taosArrayGet(pList, i);
|
||||
if (taskId == *pId) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -2143,7 +2141,7 @@ int32_t mndProcessStreamReqCheckpoint(SRpcMsg *pReq) {
|
|||
|
||||
SStreamObj *pStream = mndGetStreamObj(pMnode, req.streamId);
|
||||
if (pStream == NULL) {
|
||||
mError("failed to find the stream:0x%"PRIx64" not handle the checkpoint req", req.streamId);
|
||||
mError("failed to find the stream:0x%" PRIx64 " not handle the checkpoint req", req.streamId);
|
||||
terrno = TSDB_CODE_MND_STREAM_NOT_EXIST;
|
||||
taosThreadMutexUnlock(&execInfo.lock);
|
||||
|
||||
|
|
@ -2151,13 +2149,13 @@ int32_t mndProcessStreamReqCheckpoint(SRpcMsg *pReq) {
|
|||
}
|
||||
|
||||
int32_t numOfTasks = mndGetNumOfStreamTasks(pStream);
|
||||
SArray **pReqTaskList = (SArray**)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
|
||||
SArray **pReqTaskList = (SArray **)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
|
||||
if (pReqTaskList == NULL) {
|
||||
SArray *pList = taosArrayInit(4, sizeof(int32_t));
|
||||
doAddTaskId(pList, req.taskId, pStream->uid, numOfTasks);
|
||||
taosHashPut(execInfo.pTransferStateStreams, &req.streamId, sizeof(int64_t), &pList, sizeof(void *));
|
||||
|
||||
pReqTaskList = (SArray**)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
|
||||
pReqTaskList = (SArray **)taosHashGet(execInfo.pTransferStateStreams, &req.streamId, sizeof(req.streamId));
|
||||
} else {
|
||||
doAddTaskId(*pReqTaskList, req.taskId, pStream->uid, numOfTasks);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,12 +33,15 @@ static int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSData
|
|||
int64_t suid);
|
||||
static int32_t doBuildAndSendSubmitMsg(SVnode* pVnode, SStreamTask* pTask, SSubmitReq2* pReq, int32_t numOfBlocks);
|
||||
static int32_t buildSubmitMsgImpl(SSubmitReq2* pSubmitReq, int32_t vgId, void** pMsg, int32_t* msgLen);
|
||||
static int32_t doConvertRows(SSubmitTbData* pTableData, const STSchema* pTSchema, SSDataBlock* pDataBlock, const char* id);
|
||||
static int32_t doConvertRows(SSubmitTbData* pTableData, const STSchema* pTSchema, SSDataBlock* pDataBlock,
|
||||
const char* id);
|
||||
static int32_t doWaitForDstTableCreated(SVnode* pVnode, SStreamTask* pTask, STableSinkInfo* pTableSinkInfo,
|
||||
const char* dstTableName, int64_t* uid);
|
||||
static int32_t doPutIntoCache(SSHashObj* pSinkTableMap, STableSinkInfo* pTableSinkInfo, uint64_t groupId, const char* id);
|
||||
static int32_t doPutIntoCache(SSHashObj* pSinkTableMap, STableSinkInfo* pTableSinkInfo, uint64_t groupId,
|
||||
const char* id);
|
||||
static bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbName, int64_t suid);
|
||||
static int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName, int32_t numOfTags);
|
||||
static int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName,
|
||||
int32_t numOfTags);
|
||||
static SArray* createDefaultTagColName();
|
||||
static void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
|
||||
int64_t gid, bool newSubTableRule);
|
||||
|
|
@ -68,10 +71,7 @@ int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* p
|
|||
if (varTbName != NULL && varTbName != (void*)-1) {
|
||||
name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
|
||||
memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
|
||||
if(newSubTableRule &&
|
||||
!isAutoTableName(name) &&
|
||||
!alreadyAddGroupId(name) &&
|
||||
groupId != 0) {
|
||||
if (newSubTableRule && !isAutoTableName(name) && !alreadyAddGroupId(name) && groupId != 0) {
|
||||
buildCtbNameAddGruopId(name, groupId);
|
||||
}
|
||||
} else if (stbFullName) {
|
||||
|
|
@ -134,7 +134,7 @@ end:
|
|||
return ret;
|
||||
}
|
||||
|
||||
static bool tqGetTableInfo(SSHashObj* pTableInfoMap,uint64_t groupId, STableSinkInfo** pInfo) {
|
||||
static bool tqGetTableInfo(SSHashObj* pTableInfoMap, uint64_t groupId, STableSinkInfo** pInfo) {
|
||||
void* pVal = tSimpleHashGet(pTableInfoMap, &groupId, sizeof(uint64_t));
|
||||
if (pVal) {
|
||||
*pInfo = *(STableSinkInfo**)pVal;
|
||||
|
|
@ -149,7 +149,7 @@ static int32_t tqPutReqToQueue(SVnode* pVnode, SVCreateTbBatchReq* pReqs) {
|
|||
int32_t tlen = 0;
|
||||
encodeCreateChildTableForRPC(pReqs, TD_VID(pVnode), &buf, &tlen);
|
||||
|
||||
SRpcMsg msg = { .msgType = TDMT_VND_CREATE_TABLE, .pCont = buf, .contLen = tlen };
|
||||
SRpcMsg msg = {.msgType = TDMT_VND_CREATE_TABLE, .pCont = buf, .contLen = tlen};
|
||||
if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
|
||||
tqError("failed to put into write-queue since %s", terrstr());
|
||||
}
|
||||
|
|
@ -181,14 +181,12 @@ SArray* createDefaultTagColName() {
|
|||
void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
|
||||
int64_t gid, bool newSubTableRule) {
|
||||
if (pDataBlock->info.parTbName[0]) {
|
||||
if(newSubTableRule &&
|
||||
!isAutoTableName(pDataBlock->info.parTbName) &&
|
||||
!alreadyAddGroupId(pDataBlock->info.parTbName) &&
|
||||
gid != 0) {
|
||||
if (newSubTableRule && !isAutoTableName(pDataBlock->info.parTbName) &&
|
||||
!alreadyAddGroupId(pDataBlock->info.parTbName) && gid != 0) {
|
||||
pCreateTableReq->name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
|
||||
strcpy(pCreateTableReq->name, pDataBlock->info.parTbName);
|
||||
buildCtbNameAddGruopId(pCreateTableReq->name, gid);
|
||||
}else{
|
||||
} else {
|
||||
pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
|
||||
}
|
||||
} else {
|
||||
|
|
@ -196,13 +194,14 @@ void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDa
|
|||
}
|
||||
}
|
||||
|
||||
static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* pDataBlock, SStreamTask* pTask,
|
||||
int64_t suid) {
|
||||
static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* pDataBlock,
|
||||
SStreamTask* pTask, int64_t suid) {
|
||||
tqDebug("s-task:%s build create table msg", pTask->id.idStr);
|
||||
|
||||
STSchema* pTSchema = pTask->outputInfo.tbSink.pTSchema;
|
||||
int32_t rows = pDataBlock->info.rows;
|
||||
SArray* tagArray = taosArrayInit(4, sizeof(STagVal));;
|
||||
SArray* tagArray = taosArrayInit(4, sizeof(STagVal));
|
||||
;
|
||||
int32_t code = 0;
|
||||
|
||||
SVCreateTbBatchReq reqs = {0};
|
||||
|
|
@ -262,7 +261,8 @@ static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, S
|
|||
ASSERT(gid == *(int64_t*)pGpIdData);
|
||||
}
|
||||
|
||||
setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
|
||||
setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid,
|
||||
pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && pTask->subtableWithoutMd5 != 1);
|
||||
|
||||
taosArrayPush(reqs.pArray, pCreateTbReq);
|
||||
tqDebug("s-task:%s build create table:%s msg complete", pTask->id.idStr, pCreateTbReq->name);
|
||||
|
|
@ -274,7 +274,7 @@ static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, S
|
|||
tqError("s-task:%s failed to send create table msg", pTask->id.idStr);
|
||||
}
|
||||
|
||||
_end:
|
||||
_end:
|
||||
taosArrayDestroy(tagArray);
|
||||
taosArrayDestroyEx(crTblArray, (FDelete)tdDestroySVCreateTbReq);
|
||||
return code;
|
||||
|
|
@ -361,7 +361,8 @@ int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, c
|
|||
pExisted->aRowP = pFinal;
|
||||
|
||||
tqTrace("s-task:%s rows merged, final rows:%d, uid:%" PRId64 ", existed auto-create table:%d, new-block:%d", id,
|
||||
(int32_t)taosArrayGetSize(pFinal), pExisted->uid, (pExisted->pCreateTbReq != NULL), (pNew->pCreateTbReq != NULL));
|
||||
(int32_t)taosArrayGetSize(pFinal), pExisted->uid, (pExisted->pCreateTbReq != NULL),
|
||||
(pNew->pCreateTbReq != NULL));
|
||||
|
||||
tdDestroySVCreateTbReq(pNew->pCreateTbReq);
|
||||
taosMemoryFree(pNew->pCreateTbReq);
|
||||
|
|
@ -373,7 +374,7 @@ int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock*
|
|||
SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))};
|
||||
|
||||
int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr,
|
||||
pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
|
||||
pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && pTask->subtableWithoutMd5 != 1);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return code;
|
||||
}
|
||||
|
|
@ -416,8 +417,8 @@ bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbNam
|
|||
}
|
||||
|
||||
if (pReader->me.ctbEntry.suid != suid) {
|
||||
tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid:%" PRId64 ", actual:%" PRId64,
|
||||
vgId, ctbName, suid, pReader->me.ctbEntry.suid);
|
||||
tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid:%" PRId64 ", actual:%" PRId64, vgId,
|
||||
ctbName, suid, pReader->me.ctbEntry.suid);
|
||||
terrno = TSDB_CODE_TDB_TABLE_IN_OTHER_STABLE;
|
||||
return false;
|
||||
}
|
||||
|
|
@ -437,10 +438,10 @@ SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, in
|
|||
taosArrayClear(pTagArray);
|
||||
initCreateTableMsg(pCreateTbReq, suid, stbFullName, 1);
|
||||
|
||||
STagVal tagVal = { .cid = numOfCols, .type = TSDB_DATA_TYPE_UBIGINT, .i64 = pDataBlock->info.id.groupId};
|
||||
STagVal tagVal = {.cid = numOfCols, .type = TSDB_DATA_TYPE_UBIGINT, .i64 = pDataBlock->info.id.groupId};
|
||||
taosArrayPush(pTagArray, &tagVal);
|
||||
|
||||
tTagNew(pTagArray, 1, false, (STag**) &pCreateTbReq->ctb.pTag);
|
||||
tTagNew(pTagArray, 1, false, (STag**)&pCreateTbReq->ctb.pTag);
|
||||
|
||||
if (pCreateTbReq->ctb.pTag == NULL) {
|
||||
tdDestroySVCreateTbReq(pCreateTbReq);
|
||||
|
|
@ -513,13 +514,13 @@ int32_t buildSubmitMsgImpl(SSubmitReq2* pSubmitReq, int32_t vgId, void** pMsg, i
|
|||
}
|
||||
|
||||
int32_t tsAscendingSortFn(const void* p1, const void* p2) {
|
||||
SRow* pRow1 = *(SRow**) p1;
|
||||
SRow* pRow2 = *(SRow**) p2;
|
||||
SRow* pRow1 = *(SRow**)p1;
|
||||
SRow* pRow2 = *(SRow**)p2;
|
||||
|
||||
if (pRow1->ts == pRow2->ts) {
|
||||
return 0;
|
||||
} else {
|
||||
return pRow1->ts > pRow2->ts? 1:-1;
|
||||
return pRow1->ts > pRow2->ts ? 1 : -1;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -563,7 +564,7 @@ int32_t doConvertRows(SSubmitTbData* pTableData, const STSchema* pTSchema, SSDat
|
|||
void* colData = colDataGetData(pColData, j);
|
||||
if (IS_STR_DATA_TYPE(pCol->type)) {
|
||||
// address copy, no value
|
||||
SValue sv = (SValue){.nData = varDataLen(colData), .pData = (uint8_t*) varDataVal(colData)};
|
||||
SValue sv = (SValue){.nData = varDataLen(colData), .pData = (uint8_t*)varDataVal(colData)};
|
||||
SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, sv);
|
||||
taosArrayPush(pVals, &cv);
|
||||
} else {
|
||||
|
|
@ -666,11 +667,9 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
|
|||
if (dstTableName[0] == 0) {
|
||||
memset(dstTableName, 0, TSDB_TABLE_NAME_LEN);
|
||||
buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName);
|
||||
}else{
|
||||
if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
|
||||
!isAutoTableName(dstTableName) &&
|
||||
!alreadyAddGroupId(dstTableName) &&
|
||||
groupId != 0) {
|
||||
} else {
|
||||
if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && pTask->subtableWithoutMd5 != 1 &&
|
||||
!isAutoTableName(dstTableName) && !alreadyAddGroupId(dstTableName) && groupId != 0) {
|
||||
buildCtbNameAddGruopId(dstTableName, groupId);
|
||||
}
|
||||
}
|
||||
|
|
@ -693,7 +692,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
|
|||
tqTrace("s-task:%s cached tableInfo uid is invalid, acquire it from meta", id);
|
||||
return doWaitForDstTableCreated(pVnode, pTask, pTableSinkInfo, dstTableName, &pTableData->uid);
|
||||
} else {
|
||||
tqTrace("s-task:%s set the dstTable uid from cache:%"PRId64, id, pTableData->uid);
|
||||
tqTrace("s-task:%s set the dstTable uid from cache:%" PRId64, id, pTableData->uid);
|
||||
}
|
||||
} else {
|
||||
// The auto-create option will always set to be open for those submit messages, which arrive during the period
|
||||
|
|
@ -714,7 +713,8 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
|
|||
|
||||
pTableData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
|
||||
pTableData->pCreateTbReq =
|
||||
buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER);
|
||||
buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray,
|
||||
pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && pTask->subtableWithoutMd5 != 1);
|
||||
taosArrayDestroy(pTagArray);
|
||||
|
||||
if (pTableData->pCreateTbReq == NULL) {
|
||||
|
|
@ -746,12 +746,12 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
|
|||
return TDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t tqSetDstTableDataPayload(uint64_t suid, const STSchema *pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock,
|
||||
int32_t tqSetDstTableDataPayload(uint64_t suid, const STSchema* pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock,
|
||||
SSubmitTbData* pTableData, const char* id) {
|
||||
int32_t numOfRows = pDataBlock->info.rows;
|
||||
|
||||
tqDebug("s-task:%s sink data pipeline, build submit msg from %dth resBlock, including %d rows, dst suid:%" PRId64,
|
||||
id, blockIndex + 1, numOfRows, suid);
|
||||
tqDebug("s-task:%s sink data pipeline, build submit msg from %dth resBlock, including %d rows, dst suid:%" PRId64, id,
|
||||
blockIndex + 1, numOfRows, suid);
|
||||
char* dstTableName = pDataBlock->info.parTbName;
|
||||
|
||||
// convert all rows
|
||||
|
|
@ -767,7 +767,7 @@ int32_t tqSetDstTableDataPayload(uint64_t suid, const STSchema *pTSchema, int32_
|
|||
}
|
||||
|
||||
bool hasOnlySubmitData(const SArray* pBlocks, int32_t numOfBlocks) {
|
||||
for(int32_t i = 0; i < numOfBlocks; ++i) {
|
||||
for (int32_t i = 0; i < numOfBlocks; ++i) {
|
||||
SSDataBlock* p = taosArrayGet(pBlocks, i);
|
||||
if (p->info.type == STREAM_DELETE_RESULT || p->info.type == STREAM_CREATE_CHILD_TABLE) {
|
||||
return false;
|
||||
|
|
@ -793,7 +793,7 @@ void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data) {
|
|||
tqDebug("vgId:%d, s-task:%s write %d stream resBlock(s) into table, has delete block, submit one-by-one", vgId, id,
|
||||
numOfBlocks);
|
||||
|
||||
for(int32_t i = 0; i < numOfBlocks; ++i) {
|
||||
for (int32_t i = 0; i < numOfBlocks; ++i) {
|
||||
if (streamTaskShouldStop(pTask)) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -832,7 +832,8 @@ void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data) {
|
|||
}
|
||||
} else {
|
||||
tqDebug("vgId:%d, s-task:%s write %d stream resBlock(s) into table, merge submit msg", vgId, id, numOfBlocks);
|
||||
SHashObj* pTableIndexMap = taosHashInit(numOfBlocks, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
|
||||
SHashObj* pTableIndexMap =
|
||||
taosHashInit(numOfBlocks, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
|
||||
|
||||
SSubmitReq2 submitReq = {.aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData))};
|
||||
if (submitReq.aSubmitTbData == NULL) {
|
||||
|
|
|
|||
|
|
@ -569,6 +569,7 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
|
|||
char ctbName[TSDB_TABLE_FNAME_LEN] = {0};
|
||||
if (pDataBlock->info.parTbName[0]) {
|
||||
if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
|
||||
pTask->subtableWithoutMd5 != 1 &&
|
||||
!isAutoTableName(pDataBlock->info.parTbName) &&
|
||||
!alreadyAddGroupId(pDataBlock->info.parTbName) &&
|
||||
groupId != 0){
|
||||
|
|
|
|||
|
|
@ -80,7 +80,7 @@ static SStreamChildEpInfo* createStreamTaskEpInfo(const SStreamTask* pTask) {
|
|||
}
|
||||
|
||||
SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, SEpSet* pEpset, bool fillHistory, int64_t triggerParam,
|
||||
SArray* pTaskList, bool hasFillhistory) {
|
||||
SArray* pTaskList, bool hasFillhistory, int8_t subtableWithoutMd5) {
|
||||
SStreamTask* pTask = (SStreamTask*)taosMemoryCalloc(1, sizeof(SStreamTask));
|
||||
if (pTask == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
|
|
@ -96,6 +96,7 @@ SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, SEpSet* pEpset,
|
|||
pTask->info.taskLevel = taskLevel;
|
||||
pTask->info.fillHistory = fillHistory;
|
||||
pTask->info.triggerParam = triggerParam;
|
||||
pTask->subtableWithoutMd5 = subtableWithoutMd5;
|
||||
|
||||
pTask->status.pSM = streamCreateStateMachine(pTask);
|
||||
if (pTask->status.pSM == NULL) {
|
||||
|
|
@ -205,6 +206,7 @@ int32_t tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
|
|||
if (tEncodeCStr(pEncoder, pTask->outputInfo.shuffleDispatcher.stbFullName) < 0) return -1;
|
||||
}
|
||||
if (tEncodeI64(pEncoder, pTask->info.triggerParam) < 0) return -1;
|
||||
if (tEncodeI8(pEncoder, pTask->subtableWithoutMd5) < 0) return -1;
|
||||
if (tEncodeCStrWithLen(pEncoder, pTask->reserve, sizeof(pTask->reserve) - 1) < 0) return -1;
|
||||
|
||||
tEndEncode(pEncoder);
|
||||
|
|
@ -287,6 +289,7 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
|
|||
if (tDecodeCStrTo(pDecoder, pTask->outputInfo.shuffleDispatcher.stbFullName) < 0) return -1;
|
||||
}
|
||||
if (tDecodeI64(pDecoder, &pTask->info.triggerParam) < 0) return -1;
|
||||
if (tDecodeI8(pDecoder, &pTask->subtableWithoutMd5) < 0) return -1;
|
||||
if (tDecodeCStrTo(pDecoder, pTask->reserve) < 0) return -1;
|
||||
|
||||
tEndDecode(pDecoder);
|
||||
|
|
@ -881,9 +884,7 @@ void streamTaskResume(SStreamTask* pTask) {
|
|||
}
|
||||
}
|
||||
|
||||
bool streamTaskIsSinkTask(const SStreamTask* pTask) {
|
||||
return pTask->info.taskLevel == TASK_LEVEL__SINK;
|
||||
}
|
||||
bool streamTaskIsSinkTask(const SStreamTask* pTask) { return pTask->info.taskLevel == TASK_LEVEL__SINK; }
|
||||
|
||||
int32_t streamTaskSendCheckpointReq(SStreamTask* pTask) {
|
||||
int32_t code;
|
||||
|
|
|
|||
Loading…
Reference in New Issue