This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e18b260c0f [Feature][Connector-V2] Support multi-table IoTDB source
(#12246)
e18b260c0f is described below
commit e18b260c0f215a160d3bf85b5a04bcf71409c8c3
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Sep 13 05:47:42 2026 +0000
[Feature][Connector-V2] Support multi-table IoTDB source (#12246)
---
docs/en/connectors/source/IoTDB.md | 49 ++-
docs/zh/connectors/source/IoTDB.md | 49 ++-
.../seatunnel/iotdb/source/IoTDBSource.java | 16 +-
.../seatunnel/iotdb/source/IoTDBSourceFactory.java | 156 ++++++++-
.../seatunnel/iotdb/source/IoTDBSourceReader.java | 36 +-
.../seatunnel/iotdb/source/IoTDBSourceSplit.java | 11 +
.../iotdb/source/IoTDBSourceSplitEnumerator.java | 43 ++-
.../seatunnel/iotdb/IoTDBFactoryTest.java | 102 ++++++
.../iotdb/source/IoTDBMultiTableSourceTest.java | 365 +++++++++++++++++++++
.../connector-iotdb-e2e/pom.xml | 6 +
.../seatunnel/e2e/connector/iotdb/IoTDBIT.java | 12 +
.../connector/iotdb/IoTDBMultiTableSourceIT.java | 170 ++++++++++
.../resources/iotdb/iotdb_multi_table_source.conf | 69 ++++
13 files changed, 1061 insertions(+), 23 deletions(-)
diff --git a/docs/en/connectors/source/IoTDB.md
b/docs/en/connectors/source/IoTDB.md
index 795e010c92..4a43b2c9b6 100644
--- a/docs/en/connectors/source/IoTDB.md
+++ b/docs/en/connectors/source/IoTDB.md
@@ -24,6 +24,7 @@ The current source runs a bounded SQL query. It is suitable
for batch reads and
- [x] [column projection](../../introduction/concepts/connector-v2-features.md)
> IoTDB allows column projection using SQL query.
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
+- [x] [multiple table](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
@@ -53,13 +54,16 @@ The current source runs a bounded SQL query. It is suitable
for batch reads and
## Source Options
+Use either a root-level `sql` and `schema` for one table, or `tables_configs`
for independently configured tables. The existing single-table configuration
remains supported.
+
| Name | Type | Required | Default Value |
Description
|
|----------------------------|---------|----------|---------------|-------------------------------------------------------------------------------------------------------------------|
| node_urls | string | yes | - | IoTDB
cluster address, the format is `"host1:port"` or `"host1:port,host2:port"`
|
| username | string | yes | - | IoTDB user
username
|
| password | string | yes | - | IoTDB user
password
|
-| sql | string | yes | - | execute
sql statement
|
-| schema | config | yes | - | The data
schema. For more details, please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
|
+| sql | string | conditional | - | SQL query.
Required with `schema` when `tables_configs` is not configured.
|
+| tables_configs | array | no | - | Non-empty
list of table configurations, each containing `sql` and `schema` with a unique,
non-blank `schema.table`. |
+| schema | config | conditional | - | Required
with root-level `sql`; configure inside each entry when using `tables_configs`.
See [Schema Feature](../../introduction/concepts/schema-feature.md). |
| fetch_size | int | no | - | Number of
rows fetched from IoTDB in one request.
|
| lower_bound | long | no | - | Lower time
bound used when SeaTunnel splits the query by time.
|
| upper_bound | long | no | - | Upper time
bound used when SeaTunnel splits the query by time.
|
@@ -72,6 +76,47 @@ The current source runs a bounded SQL query. It is suitable
for batch reads and
The first field in `schema.fields` must describe the IoTDB time column. It can
be `bigint` when you want epoch milliseconds, or `timestamp` when you want a
SeaTunnel timestamp value.
+### Multiple tables
+
+All tables share the source-level connection and client options, including
`node_urls`, credentials, `fetch_size`, and `version`. Configure these options
only at source level. Each `tables_configs` entry defines its own SQL query,
schema, and optional `lower_bound`, `upper_bound`, and `num_partitions`.
Root-level SQL, schema, and time-partition options cannot be combined with
`tables_configs`.
+
+`schema.table` identifies the output table for downstream routing; it does not
change the IoTDB path in the SQL. Each schema must match its query's columns in
order, starting with the time column. Different entries can have different
field names and types.
+
+For a table with time partitions, provide all three partition options, a
positive `num_partitions`, and `lower_bound < upper_bound`. The range is
inclusive at both configured bounds and is divided into non-overlapping
partitions. At most one partition per timestamp is created. `upper_bound =
Long.MAX_VALUE` and ranges whose inclusive size exceeds `Long.MAX_VALUE` are
not supported. Without partition options, that table's SQL runs unchanged as
one split. Existing root-level partition beha [...]
+
+Per-table time partitioning supports simple SELECT projections with optional
WHERE and ALIGN BY clauses. Functions in the SELECT list, subqueries, quoted
expressions/identifiers, SQL comments, semicolons, GROUP BY, ORDER BY,
LIMIT/OFFSET, SLIMIT/SOFFSET, FILL and INTO are rejected in this mode because
adding a time predicate cannot safely preserve their semantics. Leave partition
options unset to execute such queries unchanged.
+
+```hocon
+source {
+ IoTDB {
+ node_urls = "localhost:6667"
+ username = root
+ password = root
+ tables_configs = [
+ {
+ sql = "SELECT temperature FROM root.weather.device_a"
+ lower_bound = 1
+ upper_bound = 100
+ num_partitions = 4
+ schema {
+ table = weather
+ fields {ts = bigint, temperature = float}
+ }
+ },
+ {
+ sql = "SELECT enabled FROM root.status.device_b"
+ schema {
+ table = status
+ fields {ts = bigint, enabled = boolean}
+ }
+ }
+ ]
+ }
+}
+```
+
+This remains a bounded source, not CDC. Checkpoints preserve each pending
split's table identity. Legacy single-table checkpoints remain readable with
the legacy configuration; switching a running job from single-table
configuration to `tables_configs` requires starting a new job rather than
restoring its old checkpoint.
+
When the SQL uses `align by device`, the second field normally describes the
IoTDB device name. The remaining fields must follow the same order as the
measurements returned by the SQL query.
You can use the time column as a partition key in SQL queries.
diff --git a/docs/zh/connectors/source/IoTDB.md
b/docs/zh/connectors/source/IoTDB.md
index bcbe380e03..4ed9a8ec8d 100644
--- a/docs/zh/connectors/source/IoTDB.md
+++ b/docs/zh/connectors/source/IoTDB.md
@@ -24,6 +24,7 @@ import ChangeLog from '../changelog/connector-iotdb.md';
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
> IoTDB 通过 SQL 查询支持列投影功能。
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
+- [x] [多表读取](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
@@ -53,13 +54,16 @@ import ChangeLog from '../changelog/connector-iotdb.md';
## Source 选项
+单表读取使用根级别的 `sql` 和 `schema`；多表读取使用 `tables_configs`。原有单表配置保持兼容。
+
| 名称 | 类型 | 是否必填 | 默认值 | 描述
|
|----------------------------|---------|------|-----|----------------------------------------------------------------------------------|
| node_urls | string | 是 | - | IoTDB 集群地址，格式为
`"host1:port"` 或 `"host1:port,host2:port"` |
| username | string | 是 | - | IoTDB 用户名
|
| password | string | 是 | - | IoTDB 用户密码
|
-| sql | string | 是 | - | 要执行的 SQL 查询语句
|
-| schema | config | 是 | - | 数据模式定义。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
+| sql | string | 条件必填 | - | 未配置 `tables_configs` 时，必须与
`schema` 一起配置。 |
+| tables_configs | array | 否 | - | 非空表配置列表。每项包含 `sql` 和
`schema`，且 `schema.table` 必须非空、唯一。 |
+| schema | config | 条件必填 | - | 根级别 `sql` 配置需要此项；使用
`tables_configs` 时放在每项内。参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。 |
| fetch_size | int | 否 | - | 单次获取数据量：查询时每次从 IoTDB
获取的数据量 |
| lower_bound | long | 否 | - | 时间范围下界（通过时间列进行数据分片时使用）
|
| upper_bound | long | 否 | - | 时间范围上界（通过时间列进行数据分片时使用）
|
@@ -72,6 +76,47 @@ import ChangeLog from '../changelog/connector-iotdb.md';
`schema.fields` 中的第一个字段必须对应 IoTDB 时间列。需要毫秒时间戳时可以配置为 `bigint`，需要 SeaTunnel
timestamp 值时可以配置为 `timestamp`。
+### 多表读取
+
+所有表共用 Source 级别的连接及客户端参数，包括 `node_urls`、用户名、密码、`fetch_size` 和
`version`。这些参数只能放在 Source 级别。每个 `tables_configs` 项独立配置 SQL、schema，以及可选的
`lower_bound`、`upper_bound`、`num_partitions`。不能同时配置根级别的 SQL、schema 或时间分片参数。
+
+`schema.table` 用于下游表路由，不会改变 SQL 中的 IoTDB 路径。各表可以有不同字段及类型，字段顺序必须匹配查询结果，首字段为时间列。
+
+如果启用时间分片，必须同时配置三个分片参数，`num_partitions` 必须为正数且 `lower_bound <
upper_bound`。每个表按包含两端的时间范围生成不重叠的分片，最多每个时间戳一个分片。不支持 `upper_bound =
Long.MAX_VALUE` 或包含的时间戳数量超过 `Long.MAX_VALUE` 的范围。未配置分片参数时，该表的 SQL
原样作为一个分片执行。原有根级别的分片行为不变。
+
+表级时间分片支持带可选 WHERE 和 ALIGN BY 子句的简单 SELECT 投影。此模式拒绝 SELECT
列表中的函数、子查询、带引号的表达式/标识符、SQL 注释、分号、GROUP BY、ORDER
BY、LIMIT/OFFSET、SLIMIT/SOFFSET、FILL、INTO，因为添加时间条件无法安全保留这些查询的语义。此类查询请不配置分片参数，以原始
SQL 执行。
+
+```hocon
+source {
+ IoTDB {
+ node_urls = "localhost:6667"
+ username = root
+ password = root
+ tables_configs = [
+ {
+ sql = "SELECT temperature FROM root.weather.device_a"
+ lower_bound = 1
+ upper_bound = 100
+ num_partitions = 4
+ schema {
+ table = weather
+ fields {ts = bigint, temperature = float}
+ }
+ },
+ {
+ sql = "SELECT enabled FROM root.status.device_b"
+ schema {
+ table = status
+ fields {ts = bigint, enabled = boolean}
+ }
+ }
+ ]
+ }
+}
+```
+
+多表读取仍为有界批量读取，不提供 CDC。检查点保留每个待处理分片的表标识。旧单表检查点可继续使用原有配置恢复；将运行中的单表任务改为
`tables_configs` 时，应启动新任务，不应从旧检查点恢复。
+
当 SQL 使用 `align by device` 时，第二个字段通常对应 IoTDB 设备名。后续字段需要和 SQL 返回的测点顺序保持一致。
可以使用时间列进行分区查询。
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSource.java
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSource.java
index a1d4e50865..898179c95c 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSource.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSource.java
@@ -28,6 +28,7 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.connectors.seatunnel.iotdb.state.IoTDBSourceState;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -36,11 +37,15 @@ public class IoTDBSource
SupportParallelism,
SupportColumnProjection {
- private CatalogTable catalogTable;
- private ReadonlyConfig pluginConfig;
+ private final List<CatalogTable> catalogTables;
+ private final ReadonlyConfig pluginConfig;
public IoTDBSource(CatalogTable catalogTable, ReadonlyConfig pluginConfig)
{
- this.catalogTable = catalogTable;
+ this(Collections.singletonList(catalogTable), pluginConfig);
+ }
+
+ IoTDBSource(List<CatalogTable> catalogTables, ReadonlyConfig pluginConfig)
{
+ this.catalogTables = Collections.unmodifiableList(new
ArrayList<>(catalogTables));
this.pluginConfig = pluginConfig;
}
@@ -57,8 +62,7 @@ public class IoTDBSource
@Override
public SourceReader<SeaTunnelRow, IoTDBSourceSplit> createReader(
SourceReader.Context readerContext) {
- return new IoTDBSourceReader(
- pluginConfig, readerContext,
catalogTable.getSeaTunnelRowType());
+ return new IoTDBSourceReader(pluginConfig, readerContext,
catalogTables);
}
@Override
@@ -77,6 +81,6 @@ public class IoTDBSource
@Override
public List<CatalogTable> getProducedCatalogTables() {
- return Collections.singletonList(catalogTable);
+ return catalogTables;
}
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceFactory.java
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceFactory.java
index 7f0007d06f..e0d6aa95db 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceFactory.java
@@ -17,7 +17,13 @@
package org.apache.seatunnel.connectors.seatunnel.iotdb.source;
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
@@ -32,6 +38,12 @@ import
org.apache.seatunnel.connectors.seatunnel.iotdb.config.IoTDBSourceOptions
import com.google.auto.service.AutoService;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
@AutoService(Factory.class)
public class IoTDBSourceFactory implements TableSourceFactory {
@@ -46,10 +58,18 @@ public class IoTDBSourceFactory implements
TableSourceFactory {
.required(
IoTDBSourceOptions.NODE_URLS,
IoTDBSourceOptions.USERNAME,
- IoTDBSourceOptions.PASSWORD,
+ IoTDBSourceOptions.PASSWORD)
+ .exclusive(IoTDBSourceOptions.SQL,
ConnectorCommonOptions.TABLE_CONFIGS)
+ .optional(
IoTDBSourceOptions.SQL,
- ConnectorCommonOptions.SCHEMA)
+ Conditions.extension(IoTDBSourceOptions.SQL, new
SingleTableValidator()))
+ .optional(
+ ConnectorCommonOptions.TABLE_CONFIGS,
+
Conditions.notEmpty(ConnectorCommonOptions.TABLE_CONFIGS),
+ Conditions.extension(
+ ConnectorCommonOptions.TABLE_CONFIGS, new
TableConfigsValidator()))
.optional(
+ ConnectorCommonOptions.SCHEMA,
IoTDBSourceOptions.FETCH_SIZE,
IoTDBSourceOptions.THRIFT_DEFAULT_BUFFER_SIZE,
IoTDBSourceOptions.THRIFT_MAX_FRAME_SIZE,
@@ -64,10 +84,138 @@ public class IoTDBSourceFactory implements
TableSourceFactory {
@Override
public <T, SplitT extends SourceSplit, StateT extends Serializable>
TableSource<T, SplitT, StateT>
createSource(TableSourceFactoryContext context) {
- CatalogTable catalogTable =
CatalogTableUtil.buildWithConfig(context.getOptions());
+ ConfigValidator.of(context.getOptions()).validate(optionRule());
+ List<CatalogTable> catalogTables = new ArrayList<>();
+ Set<String> tableIds = new HashSet<>();
+ for (ReadonlyConfig config : tableConfigs(context.getOptions())) {
+ CatalogTable table = CatalogTableUtil.buildWithConfig(config);
+ if (!tableIds.add(table.getTablePath().toString())) {
+ throw new OptionValidationException(
+ "Duplicate table identity in tables_configs: %s",
table.getTablePath());
+ }
+ catalogTables.add(table);
+ }
return () ->
(SeaTunnelSource<T, SplitT, StateT>)
- new IoTDBSource(catalogTable, context.getOptions());
+ new IoTDBSource(catalogTables, context.getOptions());
+ }
+
+ static List<ReadonlyConfig> tableConfigs(ReadonlyConfig config) {
+ if
(!config.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent()) {
+ return Collections.singletonList(config);
+ }
+ List<ReadonlyConfig> tables = new ArrayList<>();
+ for (Map<String, Object> entry :
config.get(ConnectorCommonOptions.TABLE_CONFIGS)) {
+ tables.add(ReadonlyConfig.fromMap(entry));
+ }
+ return tables;
+ }
+
+ static class SingleTableValidator implements ConditionExtension<String> {
+ @Override
+ public String description() {
+ return "root-level sql requires schema";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String sql) {
+ if
(!config.getOptional(ConnectorCommonOptions.SCHEMA).isPresent()) {
+ throw new OptionValidationException(
+ "'schema' must be configured with root-level 'sql'");
+ }
+ return true;
+ }
+ }
+
+ static class TableConfigsValidator implements
ConditionExtension<List<Map<String, Object>>> {
+ @Override
+ public String description() {
+ return "each tables_configs entry requires sql and a schema with a
unique table identity";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String,
Object>> entries) {
+ if (config.getOptional(ConnectorCommonOptions.SCHEMA).isPresent()
+ ||
config.getOptional(IoTDBSourceOptions.LOWER_BOUND).isPresent()
+ ||
config.getOptional(IoTDBSourceOptions.UPPER_BOUND).isPresent()
+ ||
config.getOptional(IoTDBSourceOptions.NUM_PARTITIONS).isPresent()) {
+ throw new OptionValidationException(
+ "With tables_configs, configure schema and time
partitions inside each entry");
+ }
+ Set<String> tableIds = new HashSet<>();
+ for (int i = 0; i < entries.size(); i++) {
+ ReadonlyConfig table = ReadonlyConfig.fromMap(entries.get(i));
+ for (Option<?> connectionOption :
+ new Option<?>[] {
+ IoTDBSourceOptions.NODE_URLS,
+ IoTDBSourceOptions.USERNAME,
+ IoTDBSourceOptions.PASSWORD,
+ IoTDBSourceOptions.FETCH_SIZE,
+ IoTDBSourceOptions.THRIFT_DEFAULT_BUFFER_SIZE,
+ IoTDBSourceOptions.THRIFT_MAX_FRAME_SIZE,
+ IoTDBSourceOptions.ENABLE_CACHE_LEADER,
+ IoTDBSourceOptions.VERSION
+ }) {
+ if (table.getOptional(connectionOption).isPresent()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: '%s' must be configured
at source level",
+ i, connectionOption.key());
+ }
+ }
+ String sql =
table.getOptional(IoTDBSourceOptions.SQL).orElse("");
+ Map<String, Object> schema =
+ table.getOptional(ConnectorCommonOptions.SCHEMA)
+ .orElse(Collections.emptyMap());
+ Object name = schema.get(ConnectorCommonOptions.TABLE.key());
+ if (sql.trim().isEmpty()
+ || !(name instanceof String)
+ || ((String) name).trim().isEmpty()) {
+ throw new OptionValidationException(
+ "tables_configs[%d] requires non-blank sql and
schema.table", i);
+ }
+ CatalogTable catalog = CatalogTableUtil.buildWithConfig(table);
+ if (!tableIds.add(catalog.getTablePath().toString())) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: duplicate table identity
'%s'",
+ i, catalog.getTablePath());
+ }
+ if
(table.getOptional(IoTDBSourceOptions.NUM_PARTITIONS).isPresent()
+ ||
table.getOptional(IoTDBSourceOptions.LOWER_BOUND).isPresent()
+ ||
table.getOptional(IoTDBSourceOptions.UPPER_BOUND).isPresent()) {
+ if (!sql.matches("(?is)^\\s*select\\s+.+\\s+from\\s+.+")
+ ||
sql.matches("(?is).*\\bselect\\b.*\\bselect\\b.*")
+ || sql.split("(?i)\\bfrom\\b", 2)[0].contains("(")
+ || sql.contains("--")
+ || sql.contains("/*")
+ || sql.contains("*/")
+ || sql.matches("(?s).*[\"'`;].*")
+ || sql.matches(
+
"(?is).*\\b(group\\s+by|order\\s+by|limit|offset|slimit|soffset|fill|into)\\b.*"))
{
+ throw new OptionValidationException(
+ "tables_configs[%d]: time partitioning
supports a simple SELECT with optional WHERE and ALIGN BY; use an unpartitioned
query for quoted expressions or other SQL clauses",
+ i);
+ }
+ if
(!table.getOptional(IoTDBSourceOptions.NUM_PARTITIONS).isPresent()
+ ||
!table.getOptional(IoTDBSourceOptions.LOWER_BOUND).isPresent()
+ ||
!table.getOptional(IoTDBSourceOptions.UPPER_BOUND).isPresent()
+ || table.get(IoTDBSourceOptions.NUM_PARTITIONS) <= 0
+ || table.get(IoTDBSourceOptions.LOWER_BOUND)
+ >=
table.get(IoTDBSourceOptions.UPPER_BOUND)
+ || table.get(IoTDBSourceOptions.UPPER_BOUND) ==
Long.MAX_VALUE
+ || table.get(IoTDBSourceOptions.UPPER_BOUND)
+ -
table.get(IoTDBSourceOptions.LOWER_BOUND)
+ < 0
+ || table.get(IoTDBSourceOptions.UPPER_BOUND)
+ -
table.get(IoTDBSourceOptions.LOWER_BOUND)
+ == Long.MAX_VALUE) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: provide positive
num_partitions and a valid lower_bound < upper_bound time range",
+ i);
+ }
+ }
+ }
+ return true;
+ }
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceReader.java
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceReader.java
index 444b68c573..203eaaabf7 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceReader.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceReader.java
@@ -17,10 +17,13 @@
package org.apache.seatunnel.connectors.seatunnel.iotdb.source;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.iotdb.exception.IotdbConnectorErrorCode;
@@ -38,8 +41,10 @@ import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -63,7 +68,7 @@ public class IoTDBSourceReader implements
SourceReader<SeaTunnelRow, IoTDBSource
private final SourceReader.Context context;
- private final SeaTunnelRowDeserializer deserializer;
+ private final Map<String, SeaTunnelRowDeserializer> deserializers = new
HashMap<>();
private Session session;
@@ -74,7 +79,25 @@ public class IoTDBSourceReader implements
SourceReader<SeaTunnelRow, IoTDBSource
this.conf = conf;
this.pendingSplits = new LinkedList<>();
this.context = readerContext;
- this.deserializer = new DefaultSeaTunnelRowDeserializer(rowType);
+ this.deserializers.put(null, new
DefaultSeaTunnelRowDeserializer(rowType));
+ }
+
+ IoTDBSourceReader(
+ ReadonlyConfig conf, SourceReader.Context readerContext,
List<CatalogTable> tables) {
+ this.conf = conf;
+ this.pendingSplits = new LinkedList<>();
+ this.context = readerContext;
+ if (!conf.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent()
+ && tables.size() == 1) {
+ deserializers.put(
+ null, new
DefaultSeaTunnelRowDeserializer(tables.get(0).getSeaTunnelRowType()));
+ } else {
+ for (CatalogTable table : tables) {
+ deserializers.put(
+ table.getTablePath().toString(),
+ new
DefaultSeaTunnelRowDeserializer(table.getSeaTunnelRowType()));
+ }
+ }
}
@Override
@@ -115,10 +138,19 @@ public class IoTDBSourceReader implements
SourceReader<SeaTunnelRow, IoTDBSource
}
private void read(IoTDBSourceSplit split, Collector<SeaTunnelRow> output)
throws Exception {
+ SeaTunnelRowDeserializer deserializer =
deserializers.get(split.getTableId());
+ if (deserializer == null) {
+ throw new IotdbConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "Unknown table identity in IoTDB split: " +
split.getTableId());
+ }
try (SessionDataSet dataSet =
session.executeQueryStatement(split.getQuery())) {
while (dataSet.hasNext()) {
RowRecord rowRecord = dataSet.next();
SeaTunnelRow seaTunnelRow =
deserializer.deserialize(rowRecord);
+ if (split.getTableId() != null) {
+ seaTunnelRow.setTableId(split.getTableId());
+ }
output.collect(seaTunnelRow);
}
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplit.java
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplit.java
index 6e7ac4e5b9..ae85a27416 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplit.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplit.java
@@ -31,6 +31,12 @@ public class IoTDBSourceSplit implements SourceSplit {
/** final query statement */
private final String query;
+ private final String tableId;
+
+ public String getTableId() {
+ return tableId;
+ }
+
@Override
public String splitId() {
return splitId;
@@ -41,7 +47,12 @@ public class IoTDBSourceSplit implements SourceSplit {
}
public IoTDBSourceSplit(String splitId, String query) {
+ this(splitId, query, null);
+ }
+
+ public IoTDBSourceSplit(String splitId, String query, String tableId) {
this.splitId = splitId;
this.query = query;
+ this.tableId = tableId;
}
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplitEnumerator.java
index d9bdd0e221..d7dd0674e1 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceSplitEnumerator.java
@@ -20,7 +20,9 @@ package
org.apache.seatunnel.connectors.seatunnel.iotdb.source;
import org.apache.seatunnel.shade.com.google.common.base.Strings;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import
org.apache.seatunnel.connectors.seatunnel.iotdb.exception.IotdbConnectorException;
import org.apache.seatunnel.connectors.seatunnel.iotdb.state.IoTDBSourceState;
@@ -121,11 +123,26 @@ public class IoTDBSourceSplitEnumerator
* <p>split 2: select * from test where (time >= 6 and time < 11) and (
age > 0 and age < 10 )
*/
private Set<IoTDBSourceSplit> getIotDBSplit() {
+ if
(!conf.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent()) {
+ return getIotDBSplit(conf, null);
+ }
+ Set<IoTDBSourceSplit> splits = new HashSet<>();
+ for (ReadonlyConfig tableConfig :
IoTDBSourceFactory.tableConfigs(conf)) {
+ String tableId =
+
CatalogTableUtil.buildWithConfig(tableConfig).getTablePath().toString();
+ splits.addAll(getIotDBSplit(tableConfig, tableId));
+ }
+ return splits;
+ }
+
+ private Set<IoTDBSourceSplit> getIotDBSplit(ReadonlyConfig conf, String
tableId) {
String sql = conf.get(SQL);
Set<IoTDBSourceSplit> iotDBSourceSplits = new HashSet<>();
// no need numPartitions, use one partition
if (!conf.getOptional(NUM_PARTITIONS).isPresent()) {
- iotDBSourceSplits.add(new IoTDBSourceSplit(DEFAULT_PARTITIONS,
sql));
+ iotDBSourceSplits.add(
+ new IoTDBSourceSplit(
+ tableId == null ? DEFAULT_PARTITIONS : tableId +
":0", sql, tableId));
return iotDBSourceSplits;
}
long start = conf.get(LOWER_BOUND);
@@ -134,12 +151,13 @@ public class IoTDBSourceSplitEnumerator
String sqlBase = sql;
String sqlAlign = null;
String sqlCondition = null;
- String[] sqls = sqlBase.split("(?i)" + SQL_ALIGN);
+ String[] sqls =
+ sqlBase.split(tableId == null ? "(?i)" + SQL_ALIGN :
"(?i)\\balign\\s+by\\b");
if (sqls.length > 1) {
sqlBase = sqls[0];
sqlAlign = sqls[1];
}
- sqls = sqlBase.split("(?i)" + SQL_WHERE);
+ sqls = sqlBase.split(tableId == null ? "(?i)" + SQL_WHERE :
"(?i)\\bwhere\\b");
if (sqls.length > SQL_WHERE_SPLIT_LENGTH) {
throw new IotdbConnectorException(
CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
@@ -154,9 +172,16 @@ public class IoTDBSourceSplitEnumerator
if (end - start < numPartitions) {
numPartitions = (int) (end - start);
}
+ if (tableId != null) {
+ long range = end - start + 1;
+ numPartitions = (int) Math.min(conf.get(NUM_PARTITIONS), range);
+ size = range / numPartitions;
+ remainder = range % numPartitions;
+ }
long currentStart = start;
int i = 0;
while (i < numPartitions) {
+ long partitionSize = tableId == null ? size : size + (i <
remainder ? 1 : 0);
String query =
" where ("
+ RESERVED_TIME
@@ -165,11 +190,11 @@ public class IoTDBSourceSplitEnumerator
+ " and "
+ RESERVED_TIME
+ " < "
- + (currentStart + size)
+ + (currentStart + partitionSize)
+ ") ";
i++;
- currentStart += size;
- if (i + 1 <= numPartitions) {
+ currentStart += partitionSize;
+ if (tableId == null && i + 1 <= numPartitions) {
currentStart = currentStart - remainder;
}
query = sqlBase + query;
@@ -179,7 +204,11 @@ public class IoTDBSourceSplitEnumerator
if (!Strings.isNullOrEmpty(sqlAlign)) {
query = query + " align by " + sqlAlign;
}
- iotDBSourceSplits.add(new
IoTDBSourceSplit(String.valueOf(query.hashCode()), query));
+ iotDBSourceSplits.add(
+ new IoTDBSourceSplit(
+ tableId == null ? String.valueOf(query.hashCode())
: tableId + ":" + i,
+ query,
+ tableId));
}
return iotDBSourceSplits;
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java
b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java
index 6e5ffdf3fd..a536d3f2ec 100644
---
a/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java
@@ -17,17 +17,119 @@
package org.apache.seatunnel.connectors.seatunnel.iotdb;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.iotdb.sink.IoTDBSinkFactory;
+import org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSource;
import
org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
class IoTDBFactoryTest {
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "tables_configs = []",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}, {sql = y, schema {table = \"a.\", fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, node_urls = \"other:6667\",
schema {table = a, fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, num_partitions = 2, lower_bound =
-9223372036854775808, upper_bound = -1, schema {table = a, fields {ts =
bigint}}}]",
+ "tables_configs = [{}]",
+ "tables_configs = [{sql = \" \" , schema {table = a, fields
{ts = bigint}}}]",
+ "tables_configs = [{sql = x, schema {fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}, {sql = y, schema {table = a, fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}]\nsql = root_query",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}]\nschema {fields {ts = bigint}}",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}]\nnum_partitions = 2",
+ "tables_configs = [{sql = x, num_partitions = 2, schema {table
= a, fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, num_partitions = 0, lower_bound =
1, upper_bound = 10, schema {table = a, fields {ts = bigint}}}]",
+ "tables_configs = [{sql = x, num_partitions = 2, lower_bound =
10, upper_bound = 1, schema {table = a, fields {ts = bigint}}}]",
+ "sql = x"
+ })
+ void rejectsInvalidTableConfiguration(String options) {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromConfig(
+ ConfigFactory.parseString(
+ "node_urls = \"localhost:6667\"\nusername =
root\npassword = root\n"
+ + options));
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () -> ConfigValidator.of(config).validate(new
IoTDBSourceFactory().optionRule()));
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "sql = x\nschema {fields {ts = bigint}}",
+ "tables_configs = [{sql = x, schema {table = a, fields {ts =
bigint}}}]"
+ })
+ void acceptsSingleTableAndLegacyConfiguration(String options) {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromConfig(
+ ConfigFactory.parseString(
+ "node_urls = \"localhost:6667\"\nusername =
root\npassword = root\n"
+ + options));
+ ConfigValidator.of(config).validate(new
IoTDBSourceFactory().optionRule());
+ }
+
+ @Test
+ void sourceFactoryProducesIndependentTables() {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromConfig(
+ ConfigFactory.parseString(
+ "node_urls = \"localhost:6667\"\nusername =
root\npassword = root\n"
+ + "tables_configs = ["
+ + "{sql = \"select temperature from
root.weather\", schema {table = weather, fields {ts = bigint, temperature =
float}}},"
+ + "{sql = \"select enabled from
root.status\", schema {table = status, fields {ts = bigint, enabled =
boolean}}}"
+ + "]"));
+ IoTDBSourceFactory factory = new IoTDBSourceFactory();
+ ConfigValidator.of(config).validate(factory.optionRule());
+ Object createdSource =
+ factory.createSource(
+ new TableSourceFactoryContext(config,
getClass().getClassLoader()))
+ .createSource();
+ IoTDBSource source = (IoTDBSource) createdSource;
+ Assertions.assertEquals(2, source.getProducedCatalogTables().size());
+ Assertions.assertEquals(
+ "weather",
source.getProducedCatalogTables().get(0).getTablePath().toString());
+ Assertions.assertEquals(
+ "status",
source.getProducedCatalogTables().get(1).getTablePath().toString());
+ }
+
@Test
void optionRule() {
Assertions.assertNotNull((new IoTDBSourceFactory()).optionRule());
Assertions.assertNotNull((new IoTDBSinkFactory()).optionRule());
}
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "SELECT value FROM root.test LIMIT 1",
+ "SELECT value FROM root.test -- comment",
+ "SELECT value FROM root.test /* comment */",
+ "SELECT count(value) FROM root.test",
+ "SELECT value FROM (SELECT value FROM root.test)",
+ "SELECT value FROM root.test WHERE name = 'where'",
+ "DELETE FROM root.test"
+ })
+ void partitionedTablesRejectQueriesThatCannotBeSafelySplit(String sql) {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromConfig(
+ ConfigFactory.parseString(
+ "node_urls = \"localhost:6667\"\nusername =
root\npassword = root\n"
+ + "tables_configs = [{sql = \""
+ + sql
+ + "\", lower_bound = 0, upper_bound =
10, num_partitions = 2, schema {table = a, fields {ts = bigint, value =
int}}}]"));
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () -> ConfigValidator.of(config).validate(new
IoTDBSourceFactory().optionRule()));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBMultiTableSourceTest.java
b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBMultiTableSourceTest.java
new file mode 100644
index 0000000000..75925b8959
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBMultiTableSourceTest.java
@@ -0,0 +1,365 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.iotdb.source;
+
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.iotdb.exception.IotdbConnectorException;
+import org.apache.seatunnel.connectors.seatunnel.iotdb.state.IoTDBSourceState;
+
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class IoTDBMultiTableSourceTest {
+
+ @ParameterizedTest
+ @CsvSource({"0,100,4", "1,2,20", "-10,10,4", "1,10,1"})
+ void tableTimePartitionsCoverEachTimestampExactlyOnce(long lower, long
upper, int count)
+ throws Exception {
+ IoTDBSource source =
+ source(
+ "tables_configs = [{sql = \"select value from
root.test\", lower_bound = "
+ + lower
+ + ", upper_bound = "
+ + upper
+ + ", num_partitions = "
+ + count
+ + ", schema {table = test, fields {ts =
bigint, value = int}}}]");
+ SourceSplitEnumerator.Context<IoTDBSourceSplit> context =
+ mock(SourceSplitEnumerator.Context.class);
+ when(context.currentParallelism()).thenReturn(1);
+ when(context.registeredReaders()).thenReturn(Collections.emptySet());
+ SourceSplitEnumerator<IoTDBSourceSplit, IoTDBSourceState> enumerator =
+ source.createEnumerator(context);
+ enumerator.run();
+ List<IoTDBSourceSplit> splits =
enumerator.snapshotState(1).getPendingSplit().get(0);
+ Assertions.assertEquals(Math.min(count, upper - lower + 1),
splits.size());
+ Set<Long> covered = new HashSet<>();
+ Pattern pattern = Pattern.compile("time >= (-?\\d+) and time <
(-?\\d+)");
+ for (IoTDBSourceSplit split : splits) {
+ Matcher matcher = pattern.matcher(split.getQuery());
+ Assertions.assertTrue(matcher.find());
+ long start = Long.parseLong(matcher.group(1));
+ long end = Long.parseLong(matcher.group(2));
+ for (long timestamp = start; timestamp < end; timestamp++) {
+ Assertions.assertTrue(covered.add(timestamp), "Overlapping
time partitions");
+ Assertions.assertTrue(timestamp >= lower && timestamp <=
upper);
+ }
+ }
+ Assertions.assertEquals(upper - lower + 1, covered.size());
+ }
+
+ private static final String CONNECTION =
+ "node_urls = \"localhost:6667\"\nusername = root\npassword =
root\n";
+ private static final String TABLES =
+ "tables_configs = ["
+ + "{sql = \"select temperature from root.weather\",
lower_bound = 1, upper_bound = 10, num_partitions = 2, schema {table = weather,
fields {ts = bigint, temperature = tinyint}}},"
+ + "{sql = \"select enabled from root.status\", schema
{table = status, fields {ts = bigint, enabled = boolean}}}"
+ + "]";
+
+ @Test
+ void enumeratesIndependentQueriesAndRestoresTableIdentities() throws
Exception {
+ IoTDBSource source = roundTrip(source(TABLES));
+ Assertions.assertEquals(Boundedness.BOUNDED, source.getBoundedness());
+ SourceSplitEnumerator.Context<IoTDBSourceSplit> context =
+ mock(SourceSplitEnumerator.Context.class);
+ when(context.currentParallelism()).thenReturn(2);
+ when(context.registeredReaders()).thenReturn(Collections.emptySet());
+ SourceSplitEnumerator<IoTDBSourceSplit, IoTDBSourceState> enumerator =
+ source.createEnumerator(context);
+ enumerator.run();
+ IoTDBSourceState state = roundTrip(enumerator.snapshotState(1));
+ List<IoTDBSourceSplit> pending =
+ state.getPendingSplit().values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ Assertions.assertEquals(3, pending.size());
+ Assertions.assertEquals(
+ 3,
pending.stream().map(IoTDBSourceSplit::splitId).distinct().count());
+ Assertions.assertEquals(
+ 2, pending.stream().filter(s ->
"weather".equals(s.getTableId())).count());
+ Assertions.assertTrue(
+ pending.stream()
+ .anyMatch(
+ s ->
+ s.getQuery().contains("time >= 1")
+ && s.getQuery().contains("time
< 6")));
+ Assertions.assertTrue(
+ pending.stream()
+ .anyMatch(
+ s ->
+ s.getQuery().contains("time >= 6")
+ && s.getQuery().contains("time
< 11")));
+ Assertions.assertTrue(
+ pending.stream()
+ .anyMatch(
+ s ->
+ "status".equals(s.getTableId())
+ && "select enabled from
root.status"
+
.equals(s.getQuery())));
+ List<IoTDBSourceSplit> assigned = new ArrayList<>();
+ doAnswer(
+ invocation -> {
+ assigned.addAll(invocation.getArgument(1));
+ return null;
+ })
+ .when(context)
+ .assignSplit(anyInt(), anyList());
+ when(context.registeredReaders()).thenReturn(new
HashSet<>(Arrays.asList(0, 1)));
+ SourceSplitEnumerator<IoTDBSourceSplit, IoTDBSourceState> restored =
+ source.restoreEnumerator(context, state);
+ restored.registerReader(0);
+ restored.registerReader(1);
+ restored.run();
+ Assertions.assertEquals(3, assigned.size());
+
Assertions.assertTrue(restored.snapshotState(2).getPendingSplit().isEmpty());
+ restored.addSplitsBack(Collections.singletonList(assigned.get(0)), 0);
+ Assertions.assertEquals(4, assigned.size());
+ }
+
+ @Test
+ void partitionSqlRecognizesKeywordsRatherThanFieldNames() throws Exception
{
+ IoTDBSource source =
+ source(
+ "tables_configs = [{sql = \"SELECT somewhere FROM
root.test WHERE somewhere > 0 ALIGN BY DEVICE\", lower_bound = 1, upper_bound =
10, num_partitions = 2, schema {table = a, fields {ts = bigint, device =
string, somewhere = int}}}]");
+ SourceSplitEnumerator.Context<IoTDBSourceSplit> context =
+ mock(SourceSplitEnumerator.Context.class);
+ when(context.currentParallelism()).thenReturn(1);
+ when(context.registeredReaders()).thenReturn(Collections.emptySet());
+ SourceSplitEnumerator<IoTDBSourceSplit, IoTDBSourceState> enumerator =
+ source.createEnumerator(context);
+ enumerator.run();
+ for (IoTDBSourceSplit split :
enumerator.snapshotState(1).getPendingSplit().get(0)) {
+ Assertions.assertTrue(split.getQuery().startsWith("SELECT
somewhere FROM root.test "));
+ Assertions.assertTrue(split.getQuery().contains("somewhere > 0"));
+ Assertions.assertTrue(split.getQuery().endsWith("align by
DEVICE"));
+ }
+ }
+
+ @Test
+ void readerUsesEachSchemaAndTableIdAndClosesResults() throws Exception {
+ SourceReader.Context context = mock(SourceReader.Context.class);
+ when(context.getBoundedness()).thenReturn(Boundedness.BOUNDED);
+ IoTDBSourceReader reader = (IoTDBSourceReader)
source(TABLES).createReader(context);
+ Session session = mock(Session.class);
+ setSession(reader, session);
+ RowRecord weather = new RowRecord(1);
+ weather.addField(7, TSDataType.INT32);
+ RowRecord status = new RowRecord(2);
+ status.addField(true, TSDataType.BOOLEAN);
+ SessionDataSet first = result(weather);
+ SessionDataSet second = result(status);
+ when(session.executeQueryStatement("weather query")).thenReturn(first);
+ when(session.executeQueryStatement("status query")).thenReturn(second);
+ reader.addSplits(
+ Arrays.asList(
+ new IoTDBSourceSplit("a", "weather query", "weather"),
+ new IoTDBSourceSplit("b", "status query", "status")));
+ List<IoTDBSourceSplit> checkpoint = roundTrip(reader.snapshotState(1));
+ Assertions.assertEquals("status", checkpoint.get(1).getTableId());
+ reader =
+ (IoTDBSourceReader)
+ source(
+ "tables_configs = ["
+ + "{sql = status, schema
{table = status, fields {ts = bigint, enabled = boolean}}},"
+ + "{sql = weather, schema
{table = weather, fields {ts = bigint, temperature = tinyint}}}]")
+ .createReader(context);
+ setSession(reader, session);
+ reader.addSplits(checkpoint);
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ reader.handleNoMoreSplits();
+ reader.pollNext(collector(rows));
+ Assertions.assertEquals(2, rows.size());
+ Assertions.assertEquals("weather", rows.get(0).getTableId());
+ Assertions.assertArrayEquals(new Object[] {1L, (byte) 7},
rows.get(0).getFields());
+ Assertions.assertEquals("status", rows.get(1).getTableId());
+ Assertions.assertArrayEquals(new Object[] {2L, true},
rows.get(1).getFields());
+ Assertions.assertTrue(reader.snapshotState(2).isEmpty());
+ verify(first).close();
+ verify(second).close();
+ verify(context).signalNoMoreElement();
+ reader.close();
+ verify(session).close();
+ }
+
+ @Test
+ void readerClosesResultsAfterConversionFailure() throws Exception {
+ IoTDBSourceReader reader =
+ (IoTDBSourceReader)
source(TABLES).createReader(mock(SourceReader.Context.class));
+ Session session = mock(Session.class);
+ setSession(reader, session);
+ SessionDataSet result = result(new RowRecord(1));
+ when(session.executeQueryStatement("invalid row")).thenReturn(result);
+ reader.addSplits(
+ Collections.singletonList(new IoTDBSourceSplit("a", "invalid
row", "weather")));
+ Assertions.assertThrows(
+ IotdbConnectorException.class, () ->
reader.pollNext(collector(new ArrayList<>())));
+ verify(result).close();
+ }
+
+ @Test
+ void readerRejectsUnknownOrMissingTableIdentity() throws Exception {
+ for (String tableId : Arrays.asList(null, "unknown")) {
+ IoTDBSourceReader reader =
+ (IoTDBSourceReader)
+
source(TABLES).createReader(mock(SourceReader.Context.class));
+ Session session = mock(Session.class);
+ setSession(reader, session);
+ reader.addSplits(
+ Collections.singletonList(new IoTDBSourceSplit("a",
"query", tableId)));
+ Assertions.assertThrows(
+ IotdbConnectorException.class,
+ () -> reader.pollNext(collector(new ArrayList<>())));
+ verifyNoInteractions(session);
+ }
+ }
+
+ @Test
+ void legacyReaderRejectsMultiTableCheckpoint() throws Exception {
+ IoTDBSourceReader reader =
+ (IoTDBSourceReader)
+ source(
+ "sql = x\nschema {table = weather,
fields {ts = bigint, temperature = int}}")
+
.createReader(mock(SourceReader.Context.class));
+ Session session = mock(Session.class);
+ setSession(reader, session);
+ reader.addSplits(Collections.singletonList(new IoTDBSourceSplit("a",
"query", "weather")));
+ Assertions.assertThrows(
+ IotdbConnectorException.class, () ->
reader.pollNext(collector(new ArrayList<>())));
+ verifyNoInteractions(session);
+ }
+
+ @Test
+ void restoresSplitSerializedBeforeTableIdentityWasAdded() throws Exception
{
+ // Generated from the original two-field IoTDBSourceSplit with
serialVersionUID = -1L.
+ String fixture =
+
"rO0ABXNyAEdvcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0b3JzLnNlYXR1bm5lbC5pb3RkYi5zb3VyY2UuSW9UREJTb3VyY2VTcGxpdP//////////AgACTAAFcXVlcnl0ABJMamF2YS9sYW5nL1N0cmluZztMAAdzcGxpdElkcQB+AAF4cHQAJHNlbGVjdCB0ZW1wZXJhdHVyZSBmcm9tIHJvb3QuZGV2aWNlc3QAE2xlZ2FjeS1kZXZpY2UtcmFuZ2U=";
+ IoTDBSourceSplit split;
+ try (ObjectInputStream input =
+ new ObjectInputStream(
+ new
ByteArrayInputStream(Base64.getDecoder().decode(fixture)))) {
+ split = (IoTDBSourceSplit) input.readObject();
+ }
+ Assertions.assertEquals("legacy-device-range", split.splitId());
+ Assertions.assertNull(split.getTableId());
+ IoTDBSourceReader reader =
+ (IoTDBSourceReader)
+ source("sql = x\nschema {fields {ts = bigint,
temperature = int}}")
+
.createReader(mock(SourceReader.Context.class));
+ Session session = mock(Session.class);
+ setSession(reader, session);
+ RowRecord row = new RowRecord(1);
+ row.addField(7, TSDataType.INT32);
+ SessionDataSet dataSet = result(row);
+
when(session.executeQueryStatement(split.getQuery())).thenReturn(dataSet);
+ reader.addSplits(Collections.singletonList(split));
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ reader.pollNext(collector(rows));
+ Assertions.assertArrayEquals(new Object[] {1L, 7},
rows.get(0).getFields());
+ Assertions.assertEquals(
+ new SeaTunnelRow(new Object[0]).getTableId(),
rows.get(0).getTableId());
+ }
+
+ private static IoTDBSource source(String options) {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromConfig(ConfigFactory.parseString(CONNECTION
+ options));
+ Object result =
+ new IoTDBSourceFactory()
+ .createSource(
+ new TableSourceFactoryContext(
+ config,
IoTDBMultiTableSourceTest.class.getClassLoader()))
+ .createSource();
+ return (IoTDBSource) result;
+ }
+
+ private static SessionDataSet result(RowRecord row) throws Exception {
+ SessionDataSet result = mock(SessionDataSet.class);
+ when(result.hasNext()).thenReturn(true, false);
+ when(result.next()).thenReturn(row);
+ return result;
+ }
+
+ private static void setSession(IoTDBSourceReader reader, Session session)
throws Exception {
+ Field field = IoTDBSourceReader.class.getDeclaredField("session");
+ field.setAccessible(true);
+ field.set(reader, session);
+ }
+
+ private static Collector<SeaTunnelRow> collector(List<SeaTunnelRow> rows) {
+ return new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow row) {
+ rows.add(row);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ };
+ }
+
+ private static <T> T roundTrip(T value) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
+ output.writeObject(value);
+ }
+ try (ObjectInputStream input =
+ new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray()))) {
+ return (T) input.readObject();
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/pom.xml
index 0140e38e3f..2694cd5be9 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/pom.xml
@@ -27,6 +27,12 @@
<dependencies>
<!-- SeaTunnel connectors -->
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-assert</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-fake</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBIT.java
index 768c499358..e5e16b7ba9 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBIT.java
@@ -110,6 +110,18 @@ public class IoTDBIT extends TestSuiteBase implements
TestResource {
assertDatasetEquals(testDataset, sinkDataset);
}
+ @TestTemplate
+ public void testMultiTableSource(TestContainer container) throws Exception
{
+ session.executeNonQueryStatement(
+ "INSERT INTO root.multi.weather(timestamp, temperature)
VALUES(1, 12.5)");
+ session.executeNonQueryStatement(
+ "INSERT INTO root.multi.weather(timestamp, temperature)
VALUES(2, 12.5)");
+ session.executeNonQueryStatement(
+ "INSERT INTO root.multi.status(timestamp, enabled) VALUES(3,
true)");
+ Container.ExecResult result =
container.executeJob("/iotdb/iotdb_multi_table_source.conf");
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+ }
+
private Session createSession() {
return new Session.Builder()
.host(iotdbServer.getHost())
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBMultiTableSourceIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBMultiTableSourceIT.java
new file mode 100644
index 0000000000..16990c5563
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iotdb/IoTDBMultiTableSourceIT.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.connector.iotdb;
+
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSource;
+import
org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSourceFactory;
+import org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSourceSplit;
+import org.apache.seatunnel.connectors.seatunnel.iotdb.state.IoTDBSourceState;
+
+import org.apache.iotdb.session.Session;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Exercises the source lifecycle against IoTDB independently of the engine
containers. */
+class IoTDBMultiTableSourceIT {
+ @Test
+ void readsDifferentSchemasAndNonOverlappingTimePartitions() throws
Exception {
+ try (GenericContainer<?> server =
+ new
GenericContainer<>("apache/iotdb:0.13.1-node").withExposedPorts(6667)) {
+ server.start();
+ Session session =
+ new Session.Builder()
+ .host(server.getHost())
+ .port(server.getMappedPort(6667))
+ .username("root")
+ .password("root")
+ .build();
+ try {
+ await().atMost(Duration.ofSeconds(60))
+ .ignoreExceptions()
+ .untilAsserted(session::open);
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES root.multi.weather.temperature WITH
DATATYPE=FLOAT, ENCODING=PLAIN");
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES root.multi.status.enabled WITH
DATATYPE=BOOLEAN, ENCODING=PLAIN");
+ for (int timestamp = 0; timestamp <= 100; timestamp++) {
+ session.executeNonQueryStatement(
+ "INSERT INTO root.multi.weather(timestamp,
temperature) VALUES("
+ + timestamp
+ + ", 12.5)");
+ }
+ session.executeNonQueryStatement(
+ "INSERT INTO root.multi.status(timestamp, enabled)
VALUES(3, true)");
+ String connection =
+ "node_urls = \""
+ + server.getHost()
+ + ":"
+ + server.getMappedPort(6667)
+ + "\"\nusername = root\npassword = root\n";
+ IoTDBSource source =
+ source(
+ connection
+ + "tables_configs = ["
+ + "{sql = \"SELECT temperature FROM
root.multi.weather\", lower_bound = 0, upper_bound = 100, num_partitions = 4,
schema {table = weather, fields {ts = bigint, temperature = float}}},"
+ + "{sql = \"SELECT enabled FROM
root.multi.status\", schema {table = status, fields {ts = bigint, enabled =
boolean}}}]");
+ List<SeaTunnelRow> rows = read(source);
+ List<SeaTunnelRow> weather =
+ rows.stream()
+ .filter(row ->
"weather".equals(row.getTableId()))
+ .collect(Collectors.toList());
+ List<SeaTunnelRow> status =
+ rows.stream()
+ .filter(row ->
"status".equals(row.getTableId()))
+ .collect(Collectors.toList());
+ Assertions.assertEquals(101, weather.size());
+ Set<Long> timestamps = new HashSet<>();
+ for (SeaTunnelRow row : weather) {
+ Assertions.assertTrue(timestamps.add((Long)
row.getField(0)));
+ Assertions.assertEquals(12.5F, row.getField(1));
+ }
+ Assertions.assertEquals(1, status.size());
+ Assertions.assertArrayEquals(new Object[] {3L, true},
status.get(0).getFields());
+ Assertions.assertEquals(
+ 101,
+ read(source(
+ connection
+ + "sql = \"SELECT temperature
FROM root.multi.weather\"\nschema {fields {ts = bigint, temperature = float}}"))
+ .size());
+ } finally {
+ session.close();
+ }
+ }
+ }
+
+ private static IoTDBSource source(String text) {
+ ReadonlyConfig config =
ReadonlyConfig.fromConfig(ConfigFactory.parseString(text));
+ Object result =
+ new IoTDBSourceFactory()
+ .createSource(
+ new TableSourceFactoryContext(
+ config,
IoTDBMultiTableSourceIT.class.getClassLoader()))
+ .createSource();
+ return (IoTDBSource) result;
+ }
+
+ private static List<SeaTunnelRow> read(IoTDBSource source) throws
Exception {
+ SourceSplitEnumerator.Context<IoTDBSourceSplit> enumeratorContext =
+ mock(SourceSplitEnumerator.Context.class);
+ when(enumeratorContext.currentParallelism()).thenReturn(2);
+
when(enumeratorContext.registeredReaders()).thenReturn(Collections.emptySet());
+ SourceSplitEnumerator<IoTDBSourceSplit, IoTDBSourceState> enumerator =
+ source.createEnumerator(enumeratorContext);
+ enumerator.run();
+ IoTDBSourceState state = enumerator.snapshotState(1);
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ Collector<SeaTunnelRow> output =
+ new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow row) {
+ rows.add(row);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ };
+ for (List<IoTDBSourceSplit> splits : state.getPendingSplit().values())
{
+ SourceReader.Context readerContext =
mock(SourceReader.Context.class);
+
when(readerContext.getBoundedness()).thenReturn(Boundedness.BOUNDED);
+ try (SourceReader<SeaTunnelRow, IoTDBSourceSplit> reader =
+ source.createReader(readerContext)) {
+ reader.open();
+ reader.addSplits(splits);
+ reader.handleNoMoreSplits();
+ reader.pollNext(output);
+ }
+ }
+ enumerator.close();
+ return rows;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/resources/iotdb/iotdb_multi_table_source.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/resources/iotdb/iotdb_multi_table_source.conf
new file mode 100644
index 0000000000..83d903ee7c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-e2e/src/test/resources/iotdb/iotdb_multi_table_source.conf
@@ -0,0 +1,69 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ IoTDB {
+ node_urls = "flink_e2e_iotdb_sink:6667"
+ username = root
+ password = root
+ tables_configs = [
+ {
+ sql = "SELECT temperature FROM root.multi.weather"
+ lower_bound = 1
+ upper_bound = 10
+ num_partitions = 4
+ schema {
+ table = weather
+ fields {ts = bigint, temperature = float}
+ }
+ },
+ {
+ sql = "SELECT enabled FROM root.multi.status"
+ schema {
+ table = status
+ fields {ts = bigint, enabled = boolean}
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ parallelism = 1
+ rules {
+ table-names = [weather, status]
+ tables_configs = [
+ {
+ table_path = weather
+ row_rules = [{rule_type = MIN_ROW, rule_value = 2}, {rule_type =
MAX_ROW, rule_value = 2}]
+ field_rules = [{field_name = temperature, field_type = float,
field_value = [{equals_to = 12.5}]}]
+ },
+ {
+ table_path = status
+ row_rules = [{rule_type = MIN_ROW, rule_value = 1}, {rule_type =
MAX_ROW, rule_value = 1}]
+ field_rules = [{field_name = enabled, field_type = boolean,
field_value = [{equals_to = true}]}]
+ }
+ ]
+ }
+ }
+}
