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 0c20c86ef8 [Feature][Connector-V2][AmazonDynamoDB] Support multi-table
source reads (#12665)
0c20c86ef8 is described below
commit 0c20c86ef8d84ba2b65bbda8784d8ec8e17b7e20
Author: Goutam Adwant <[email protected]>
AuthorDate: Fri Oct 9 03:25:33 2026 +0000
[Feature][Connector-V2][AmazonDynamoDB] Support multi-table source reads
(#12665)
---
docs/en/connectors/source/AmazonDynamoDB.md | 81 ++-
docs/zh/connectors/source/AmazonDynamoDB.md | 74 ++-
.../source/AmazonDynamoDBSource.java | 31 +-
.../source/AmazonDynamoDBSourceFactory.java | 192 +++++++-
.../source/AmazonDynamoDBSourceReader.java | 61 ++-
.../source/AmazonDynamoDBSourceSplit.java | 8 +-
.../AmazonDynamoDBSourceSplitEnumerator.java | 32 +-
...ceSplit.java => AmazonDynamoDBSourceTable.java} | 31 +-
.../source/AmazonDynamoDBMultiTableSourceTest.java | 542 +++++++++++++++++++++
.../source/AmazonDynamoDBSourceReaderTest.java | 149 ++++++
.../connector-amazondynamodb-e2e/pom.xml | 6 +
.../connector/amazondynamodb/AmazondynamodbIT.java | 19 +
.../amazondynamodbIT_multi_table_source.conf | 104 ++++
13 files changed, 1269 insertions(+), 61 deletions(-)
diff --git a/docs/en/connectors/source/AmazonDynamoDB.md
b/docs/en/connectors/source/AmazonDynamoDB.md
index 5b7d2422cf..f9c76e3b57 100644
--- a/docs/en/connectors/source/AmazonDynamoDB.md
+++ b/docs/en/connectors/source/AmazonDynamoDB.md
@@ -12,6 +12,8 @@ The connector is a batch source. DynamoDB does not expose
field types in the sam
This source reads the current table data with scan requests. It does not read
DynamoDB Streams or CDC change events.
+Use `tables_configs` to read several DynamoDB tables with different schemas
through one source.
+
## Supported Engines
> Spark<br/>
@@ -26,6 +28,7 @@ This source reads the current table data with scan requests.
It does not read Dy
- [ ] [column projection](../../introduction/concepts/connector-v2-features.md)
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+- [x] [multi-table](../../introduction/concepts/connector-v2-features.md)
## Options
@@ -35,8 +38,9 @@ This source reads the current table data with scan requests.
It does not read Dy
| region | string | yes | - | AWS region of
the DynamoDB service. |
| access_key_id | string | yes | - | AWS access key
ID. |
| secret_access_key | string | yes | - | AWS secret
access key. |
-| table | string | yes | - | DynamoDB table
name to scan. |
-| schema | config | yes | - | SeaTunnel fields
to read from DynamoDB items. |
+| table | string | no | - | Required in
single-table mode; mutually exclusive with `tables_configs`. |
+| schema | config | no | - | Required in
single-table mode; define it per entry in multi-table mode. |
+| tables_configs | list | no | - | Tables and
schemas to read in multi-table mode; see below. |
| scan_item_limit | int | no | 1 | Maximum items
returned by each scan request. |
| parallel_scan_threads | int | no | 2 | Number of
logical segments for parallel scan. |
| common-options | object | no | - | Source plugin
common parameters. |
@@ -61,7 +65,7 @@ The AWS secret access key used to connect to DynamoDB.
### table [string]
-The DynamoDB table name to scan.
+The DynamoDB table name to scan in single-table mode.
### schema [config]
@@ -91,6 +95,26 @@ schema = {
For more schema syntax, see [Schema
Feature](../../introduction/concepts/schema-feature.md).
+### tables_configs [list]
+
+An alternative to the root-level `table` and `schema` for reading several
tables. Each entry requires:
+
+- `table`: the DynamoDB table name to scan.
+- `schema`: the SeaTunnel fields to read from that table. `schema.table` sets
the output table
+ identity; when it is not set, the DynamoDB table name is used. A dot in the
name is read as a
+ `database.table` separator, so set `schema.table` when the DynamoDB table
name contains dots.
+
+An entry can also set `scan_item_limit` and `parallel_scan_threads`. When an
entry does not set
+them, the root-level values (or their defaults) are used. Each table is
scanned with its own
+`parallel_scan_threads` segments, and each row carries the identity of the
table it was read from.
+
+All entries share the root connection options `url`, `region`, `access_key_id`
and
+`secret_access_key`; connection options inside an entry are rejected. A
root-level `table` or
+`schema`, an empty list, an entry without `table` or `schema`, unsupported
entry options and
+duplicate output table identities are rejected before the job starts. Keep the
table identities
+stable when restoring a checkpoint; switching between single-table and
multi-table mode requires a
+fresh job.
+
### scan_item_limit [int]
The maximum number of items returned by each DynamoDB scan request.
@@ -139,6 +163,8 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
## Task Example
+### Read One Table
+
The following example reads rows from `source_table` and writes them to
`sink_table`.
```hocon
@@ -191,6 +217,55 @@ sink {
}
```
+### Read Multiple Tables
+
+The following example reads `orders` and `customers` with different schemas.
Rows from `customers`
+are routed downstream as `crm.customers`.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ AmazonDynamoDB {
+ url = "http://127.0.0.1:8000";
+ region = "us-east-1"
+ access_key_id = "dummy-key"
+ secret_access_key = "dummy-secret"
+ parallel_scan_threads = 2
+ tables_configs = [
+ {
+ table = "orders"
+ parallel_scan_threads = 4
+ schema = {
+ fields {
+ id = string
+ amount = int
+ }
+ }
+ },
+ {
+ table = "customers"
+ schema = {
+ table = "crm.customers"
+ fields {
+ id = string
+ name = string
+ vip = boolean
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/source/AmazonDynamoDB.md
b/docs/zh/connectors/source/AmazonDynamoDB.md
index 44839930bb..593444dd79 100644
--- a/docs/zh/connectors/source/AmazonDynamoDB.md
+++ b/docs/zh/connectors/source/AmazonDynamoDB.md
@@ -12,6 +12,8 @@ Amazon DynamoDB 源连接器通过 DynamoDB scan 请求读取已有表中的数
该源连接器使用 scan 请求读取表中当前已有的数据，不读取 DynamoDB Streams 或 CDC 变更事件。
+使用 `tables_configs` 可以在一个 source 中读取多张 schema 不同的 DynamoDB 表。
+
## 支持的引擎
> Spark<br/>
@@ -26,6 +28,7 @@ Amazon DynamoDB 源连接器通过 DynamoDB scan 请求读取已有表中的数
- [ ] [列投影](../../introduction/concepts/connector-v2-features.md)
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
+- [x] [多表读取](../../introduction/concepts/connector-v2-features.md)
## 选项
@@ -35,8 +38,9 @@ Amazon DynamoDB 源连接器通过 DynamoDB scan 请求读取已有表中的数
| region | string | 是 | - | DynamoDB 所在的 AWS 区域。 |
| access_key_id | string | 是 | - | AWS access key ID。 |
| secret_access_key | string | 是 | - | AWS secret access key。 |
-| table | string | 是 | - | 要扫描的 DynamoDB 表名。 |
-| schema | config | 是 | - | 要读取的 SeaTunnel 字段。 |
+| table | string | 否 | - | 单表模式必填，与 `tables_configs`
互斥。 |
+| schema | config | 否 | - | 单表模式必填；多表模式需要在每个条目中配置。 |
+| tables_configs | list | 否 | - | 多表模式要读取的表及 schema，详见下文。 |
| scan_item_limit | int | 否 | 1 | 每次 scan 请求返回的最大 item 数。 |
| parallel_scan_threads | int | 否 | 2 | parallel scan 的逻辑分片数。 |
| common-options | object | 否 | - | 源插件通用参数。 |
@@ -61,7 +65,7 @@ DynamoDB 所在的 AWS 区域，例如 `us-east-1`。
### table [string]
-要扫描的 DynamoDB 表名。
+单表模式下要扫描的 DynamoDB 表名。
### schema [config]
@@ -91,6 +95,20 @@ schema = {
更多 schema 写法请参考 [Schema 特性](../../introduction/concepts/schema-feature.md)。
+### tables_configs [list]
+
+用于替代根级别的 `table` 和 `schema`，读取多张表。每个条目需要配置：
+
+- `table`：要扫描的 DynamoDB 表名。
+- `schema`：从该表读取的 SeaTunnel 字段。`schema.table` 用于设置输出表标识；未配置时使用 DynamoDB
表名。表名中的点会被当作 `database.table` 的分隔符，因此当 DynamoDB 表名包含点时请设置 `schema.table`。
+
+每个条目还可以配置 `scan_item_limit` 和 `parallel_scan_threads`。条目中未配置时，使用根级别的值（或其默认值）。
+每张表按自身的 `parallel_scan_threads` 拆分 scan 分片，每行数据都会带上其来源表的标识。
+
+所有条目共享根级别的连接选项 `url`、`region`、`access_key_id` 和
`secret_access_key`，条目内的连接选项会被拒绝。
+根级别的 `table` 或 `schema`、空列表、缺少 `table` 或 `schema`
的条目、不支持的条目选项以及重复的输出表标识，都会在作业启动前报错。
+恢复 checkpoint 时请保持表标识不变；在单表模式和多表模式之间切换需要启动新作业。
+
### scan_item_limit [int]
每次 DynamoDB scan 请求最多返回的 item 数量。
@@ -139,6 +157,8 @@ DynamoDB parallel scan 使用的逻辑分片数量。
## 任务示例
+### 读取单张表
+
下面的示例从 `source_table` 读取数据，并写入 `sink_table`。
```hocon
@@ -191,6 +211,54 @@ sink {
}
```
+### 读取多张表
+
+下面的示例读取 schema 不同的 `orders` 和 `customers` 两张表。`customers` 的数据会以
`crm.customers` 作为表标识发送到下游。
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ AmazonDynamoDB {
+ url = "http://127.0.0.1:8000";
+ region = "us-east-1"
+ access_key_id = "dummy-key"
+ secret_access_key = "dummy-secret"
+ parallel_scan_threads = 2
+ tables_configs = [
+ {
+ table = "orders"
+ parallel_scan_threads = 4
+ schema = {
+ fields {
+ id = string
+ amount = int
+ }
+ }
+ },
+ {
+ table = "customers"
+ schema = {
+ table = "crm.customers"
+ fields {
+ id = string
+ name = string
+ vip = boolean
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
## 变更日志
<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSource.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSource.java
index 12f7f8435d..e6ff3791c8 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSource.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSource.java
@@ -29,6 +29,7 @@ import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDyn
import lombok.extern.slf4j.Slf4j;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -39,13 +40,27 @@ public class AmazonDynamoDBSource
SupportParallelism,
SupportColumnProjection {
- private AmazonDynamoDBConfig amazondynamodbConfig;
- private CatalogTable catalogTable;
+ private final AmazonDynamoDBConfig amazondynamodbConfig;
+ private final List<CatalogTable> catalogTables;
+ private final List<AmazonDynamoDBSourceTable> tables;
public AmazonDynamoDBSource(
AmazonDynamoDBConfig amazondynamodbConfig, CatalogTable
catalogTable) {
+ this(
+ amazondynamodbConfig,
+ Collections.singletonList(catalogTable),
+ Collections.singletonList(
+ new AmazonDynamoDBSourceTable(
+ null, amazondynamodbConfig,
catalogTable.getSeaTunnelRowType())));
+ }
+
+ AmazonDynamoDBSource(
+ AmazonDynamoDBConfig amazondynamodbConfig,
+ List<CatalogTable> catalogTables,
+ List<AmazonDynamoDBSourceTable> tables) {
this.amazondynamodbConfig = amazondynamodbConfig;
- this.catalogTable = catalogTable;
+ this.catalogTables = Collections.unmodifiableList(new
ArrayList<>(catalogTables));
+ this.tables = Collections.unmodifiableList(new ArrayList<>(tables));
}
@Override
@@ -60,7 +75,7 @@ public class AmazonDynamoDBSource
@Override
public List<CatalogTable> getProducedCatalogTables() {
- return Collections.singletonList(catalogTable);
+ return new ArrayList<>(catalogTables);
}
@Override
@@ -68,7 +83,7 @@ public class AmazonDynamoDBSource
createEnumerator(
SourceSplitEnumerator.Context<AmazonDynamoDBSourceSplit>
enumeratorContext)
throws Exception {
- return new AmazonDynamoDBSourceSplitEnumerator(enumeratorContext,
amazondynamodbConfig);
+ return new AmazonDynamoDBSourceSplitEnumerator(enumeratorContext,
null, tables);
}
@Override
@@ -77,14 +92,12 @@ public class AmazonDynamoDBSource
SourceSplitEnumerator.Context<AmazonDynamoDBSourceSplit>
enumeratorContext,
AmazonDynamoDBSourceState checkpointState)
throws Exception {
- return new AmazonDynamoDBSourceSplitEnumerator(
- enumeratorContext, amazondynamodbConfig, checkpointState);
+ return new AmazonDynamoDBSourceSplitEnumerator(enumeratorContext,
checkpointState, tables);
}
@Override
public SourceReader<SeaTunnelRow, AmazonDynamoDBSourceSplit> createReader(
SourceReader.Context readerContext) throws Exception {
- return new AmazonDynamoDBSourceReader(
- readerContext, amazondynamodbConfig,
catalogTable.getSeaTunnelRowType());
+ return new AmazonDynamoDBSourceReader(readerContext,
amazondynamodbConfig, tables);
}
}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java
index e0ad7db15d..bffdb06970 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java
@@ -17,9 +17,18 @@
package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.source;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+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.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;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
@@ -30,8 +39,16 @@ import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDyn
import com.google.auto.service.AutoService;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import static org.apache.seatunnel.api.options.ConnectorCommonOptions.SCHEMA;
+import static
org.apache.seatunnel.api.options.ConnectorCommonOptions.TABLE_CONFIGS;
import static
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBSourceOptions.ACCESS_KEY_ID;
import static
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBSourceOptions.PARALLEL_SCAN_THREADS;
import static
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBSourceOptions.REGION;
@@ -42,6 +59,18 @@ import static
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.Am
@AutoService(Factory.class)
public class AmazonDynamoDBSourceFactory implements TableSourceFactory {
+
+ private static final List<Option<?>> CONNECTION_OPTIONS =
+ Arrays.asList(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY);
+
+ private static final Set<String> TABLE_OPTION_KEYS =
+ new HashSet<>(
+ Arrays.asList(
+ TABLE.key(),
+ SCHEMA.key(),
+ SCAN_ITEM_LIMIT.key(),
+ PARALLEL_SCAN_THREADS.key()));
+
@Override
public String factoryIdentifier() {
return "AmazonDynamoDB";
@@ -50,23 +79,172 @@ public class AmazonDynamoDBSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY,
TABLE, SCHEMA)
- .optional(SCAN_ITEM_LIMIT, PARALLEL_SCAN_THREADS)
+ .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY)
+ .exclusive(TABLE, TABLE_CONFIGS)
+ .optional(TABLE, Conditions.extension(TABLE, new
SingleTableValidator()))
+ .optional(
+ TABLE_CONFIGS,
+ Conditions.notEmpty(TABLE_CONFIGS),
+ Conditions.extension(TABLE_CONFIGS, new
TableConfigsValidator()))
+ .optional(SCHEMA, SCAN_ITEM_LIMIT, PARALLEL_SCAN_THREADS)
.build();
}
@Override
public <T, SplitT extends SourceSplit, StateT extends Serializable>
TableSource<T, SplitT, StateT>
createSource(TableSourceFactoryContext context) {
- return () ->
- (SeaTunnelSource<T, SplitT, StateT>)
- new AmazonDynamoDBSource(
- new AmazonDynamoDBConfig(context.getOptions()),
-
CatalogTableUtil.buildWithConfig(context.getOptions()));
+ return () -> {
+ ReadonlyConfig config = context.getOptions();
+ if (config.getOptional(TABLE_CONFIGS).isPresent()) {
+ return (SeaTunnelSource<T, SplitT, StateT>)
createMultiTableSource(config);
+ }
+ return (SeaTunnelSource<T, SplitT, StateT>)
+ new AmazonDynamoDBSource(
+ new AmazonDynamoDBConfig(config),
+ CatalogTableUtil.buildWithConfig(config));
+ };
}
@Override
public Class<? extends SeaTunnelSource> getSourceClass() {
return AmazonDynamoDBSource.class;
}
+
+ /**
+ * Builds one table per tables_configs entry. Connection options come from
the source level;
+ * scan_item_limit and parallel_scan_threads fall back to the source level
when an entry does
+ * not set them. The table identity is schema.table, or the DynamoDB table
name when the schema
+ * does not name the table.
+ */
+ static AmazonDynamoDBSource createMultiTableSource(ReadonlyConfig config) {
+ if (config.getOptional(SCHEMA).isPresent()) {
+ throw new OptionValidationException(
+ "root-level 'schema' cannot be used with
'tables_configs'");
+ }
+ List<Map<String, Object>> entries = config.get(TABLE_CONFIGS);
+ if (entries.isEmpty()) {
+ throw new OptionValidationException("'tables_configs' must not be
empty");
+ }
+ Config sharedConfig =
config.toConfig().withoutPath(TABLE_CONFIGS.key());
+ List<CatalogTable> catalogTables = new ArrayList<>(entries.size());
+ List<AmazonDynamoDBSourceTable> tables = new
ArrayList<>(entries.size());
+ Set<String> tableIds = new HashSet<>();
+ for (int i = 0; i < entries.size(); i++) {
+ try {
+ Map<String, Object> entry = entries.get(i);
+ for (Option<?> option : CONNECTION_OPTIONS) {
+ if (entry.containsKey(option.key())) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: '%s' must be configured
at source level",
+ i, option.key());
+ }
+ }
+ for (String key : entry.keySet()) {
+ if (!TABLE_OPTION_KEYS.contains(key)) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: unsupported table option
'%s'", i, key);
+ }
+ }
+ Object tableValue = entry.get(TABLE.key());
+ // An unquoted all-digit table name is parsed as a number.
+ String tableName =
+ tableValue instanceof String || tableValue instanceof
Number
+ ? tableValue.toString()
+ : null;
+ if (tableName == null || tableName.trim().isEmpty()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'table' must be configured
and non-blank", i);
+ }
+ Object schemaValue = entry.get(SCHEMA.key());
+ if (!(schemaValue instanceof Map) || ((Map<?, ?>)
schemaValue).isEmpty()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'schema' must be configured
and non-empty", i);
+ }
+ Map<String, Object> schema = new LinkedHashMap<>((Map<String,
Object>) schemaValue);
+ Object schemaTable =
schema.get(ConnectorCommonOptions.TABLE.key());
+ if (schemaTable == null
+ || (schemaTable instanceof String
+ && ((String) schemaTable).trim().isEmpty())) {
+ schema.put(ConnectorCommonOptions.TABLE.key(), tableName);
+ }
+ Map<String, Object> tableEntry = new LinkedHashMap<>(entry);
+ tableEntry.put(TABLE.key(), tableName);
+ tableEntry.put(SCHEMA.key(), schema);
+ ReadonlyConfig tableConfig =
+ ReadonlyConfig.fromConfig(
+ ReadonlyConfig.fromMap(tableEntry)
+ .toConfig()
+ .withFallback(sharedConfig));
+
+ CatalogTable catalogTable;
+ try {
+ catalogTable =
CatalogTableUtil.buildWithConfig(tableConfig);
+ } catch (RuntimeException e) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: invalid 'schema' (set
'schema.table' when the table name is not a valid table path): %s",
+ i, e.getMessage());
+ }
+ AmazonDynamoDBConfig tableReadConfig = new
AmazonDynamoDBConfig(tableConfig);
+ if (tableReadConfig.getScanItemLimit() <= 0
+ || tableReadConfig.getParallelScanThreads() <= 0) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: '%s' and '%s' must be
positive",
+ i, SCAN_ITEM_LIMIT.key(),
PARALLEL_SCAN_THREADS.key());
+ }
+ String tableId =
catalogTable.getTableId().toTablePath().toString();
+ if (!tableIds.add(tableId)) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: duplicate table identity
'%s'", i, tableId);
+ }
+ catalogTables.add(catalogTable);
+ tables.add(
+ new AmazonDynamoDBSourceTable(
+ tableId, tableReadConfig,
catalogTable.getSeaTunnelRowType()));
+ } catch (OptionValidationException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: invalid table entry: %s", i,
e.getMessage());
+ }
+ }
+ return new AmazonDynamoDBSource(new AmazonDynamoDBConfig(config),
catalogTables, tables);
+ }
+
+ static class SingleTableValidator implements ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "'schema' must be configured when using a root-level
'table'";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String table)
+ throws OptionValidationException {
+ Map<String, Object> schema =
config.getOptional(SCHEMA).orElse(null);
+ if (schema == null || schema.isEmpty()) {
+ throw new OptionValidationException(
+ "'schema' must be configured when using a root-level
'table'");
+ }
+ return true;
+ }
+ }
+
+ static class TableConfigsValidator implements
ConditionExtension<List<Map<String, Object>>> {
+
+ @Override
+ public String description() {
+ return "each 'tables_configs' entry must contain a non-blank
'table' and a 'schema' with a unique table identity";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String,
Object>> entries)
+ throws OptionValidationException {
+ if (entries == null || entries.isEmpty()) {
+ // reported by Conditions.notEmpty
+ return true;
+ }
+ createMultiTableSource(config);
+ return true;
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
index a9e72d9977..ec886f3fbd 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReader.java
@@ -17,11 +17,13 @@
package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.source;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.source.SourceReader;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
+import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.exception.AmazonDynamoDBConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.DefaultSeaTunnelRowDeserializer;
import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.serialize.SeaTunnelRowDeserializer;
@@ -35,7 +37,10 @@ import
software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
import java.net.URI;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
@@ -47,7 +52,8 @@ public class AmazonDynamoDBSourceReader
protected DynamoDbClient dynamoDbClient;
protected SourceReader.Context context;
protected AmazonDynamoDBConfig amazondynamodbConfig;
- protected SeaTunnelRowDeserializer seaTunnelRowDeserializer;
+ private final Map<String, AmazonDynamoDBSourceTable> tables = new
HashMap<>();
+ private final Map<String, SeaTunnelRowDeserializer> deserializers = new
HashMap<>();
Queue<AmazonDynamoDBSourceSplit> pendingSplits = new
ConcurrentLinkedDeque<>();
private volatile boolean noMoreSplit;
@@ -56,9 +62,26 @@ public class AmazonDynamoDBSourceReader
SourceReader.Context context,
AmazonDynamoDBConfig amazondynamodbConfig,
SeaTunnelRowType typeInfo) {
+ this(
+ context,
+ amazondynamodbConfig,
+ Collections.singletonList(
+ new AmazonDynamoDBSourceTable(null,
amazondynamodbConfig, typeInfo)));
+ }
+
+ AmazonDynamoDBSourceReader(
+ SourceReader.Context context,
+ AmazonDynamoDBConfig amazondynamodbConfig,
+ List<AmazonDynamoDBSourceTable> tables) {
this.context = context;
this.amazondynamodbConfig = amazondynamodbConfig;
- this.seaTunnelRowDeserializer = new
DefaultSeaTunnelRowDeserializer(typeInfo);
+ // A single-table source is keyed by null, which also matches splits
restored from
+ // checkpoints taken before splits carried a table identity.
+ for (AmazonDynamoDBSourceTable table : tables) {
+ this.tables.put(table.getTableId(), table);
+ this.deserializers.put(
+ table.getTableId(), new
DefaultSeaTunnelRowDeserializer(table.getRowType()));
+ }
}
@Override
@@ -121,24 +144,34 @@ public class AmazonDynamoDBSourceReader
}
private void read(AmazonDynamoDBSourceSplit split, Collector<SeaTunnelRow>
output) {
- ScanIterable scan;
+ String tableId = split.getTableId();
+ AmazonDynamoDBSourceTable table = tables.get(tableId);
+ if (table == null) {
+ throw new AmazonDynamoDBConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ tableId == null
+ ? "Cannot restore a single-table AmazonDynamoDB
split with a tables_configs configuration"
+ : "Unknown table identity in AmazonDynamoDB split:
" + tableId);
+ }
+ SeaTunnelRowDeserializer deserializer = deserializers.get(tableId);
ScanRequest scanRequest =
ScanRequest.builder()
- .tableName(amazondynamodbConfig.getTable())
+ .tableName(table.getConfig().getTable())
.limit(split.getItemCount())
.segment(split.getSplitId())
.totalSegments(split.getTotalSegments())
.build();
- scan = dynamoDbClient.scanPaginator(scanRequest);
- do {
-
- scan.items()
- .forEach(
- item -> {
-
output.collect(seaTunnelRowDeserializer.deserialize(item));
- });
-
- } while (scan.iterator().hasNext() && !noMoreSplit);
+ // The paginator already follows every page of the segment, so iterate
it only once.
+ ScanIterable scan = dynamoDbClient.scanPaginator(scanRequest);
+ scan.items()
+ .forEach(
+ item -> {
+ SeaTunnelRow row = deserializer.deserialize(item);
+ if (tableId != null) {
+ row.setTableId(tableId);
+ }
+ output.collect(row);
+ });
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
index 825cc3666d..7e1fe36105 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
@@ -32,9 +32,15 @@ public class AmazonDynamoDBSourceSplit implements
SourceSplit {
private Integer splitId;
private Integer totalSegments;
private Integer itemCount;
+ /** Table identity for tables_configs; null for single-table splits and
older checkpoints. */
+ private String tableId;
+
+ public AmazonDynamoDBSourceSplit(Integer splitId, Integer totalSegments,
Integer itemCount) {
+ this(splitId, totalSegments, itemCount, null);
+ }
@Override
public String splitId() {
- return splitId.toString();
+ return tableId == null ? splitId.toString() : tableId + ":" + splitId;
}
}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplitEnumerator.java
index 7ba9f55f1c..29559fd35f 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplitEnumerator.java
@@ -43,7 +43,7 @@ public class AmazonDynamoDBSourceSplitEnumerator
private final SourceSplitEnumerator.Context<AmazonDynamoDBSourceSplit>
enumeratorContext;
private final Map<Integer, List<AmazonDynamoDBSourceSplit>> pendingSplits;
- private final AmazonDynamoDBConfig amazonDynamoDBConfig;
+ private final List<AmazonDynamoDBSourceTable> tables;
private final AtomicInteger assignCount = new AtomicInteger(0);
private final Object stateLock = new Object();
@@ -59,8 +59,19 @@ public class AmazonDynamoDBSourceSplitEnumerator
Context<AmazonDynamoDBSourceSplit> enumeratorContext,
AmazonDynamoDBConfig amazonDynamoDBConfig,
AmazonDynamoDBSourceState sourceState) {
+ this(
+ enumeratorContext,
+ sourceState,
+ Collections.singletonList(
+ new AmazonDynamoDBSourceTable(null,
amazonDynamoDBConfig, null)));
+ }
+
+ AmazonDynamoDBSourceSplitEnumerator(
+ Context<AmazonDynamoDBSourceSplit> enumeratorContext,
+ AmazonDynamoDBSourceState sourceState,
+ List<AmazonDynamoDBSourceTable> tables) {
this.enumeratorContext = enumeratorContext;
- this.amazonDynamoDBConfig = amazonDynamoDBConfig;
+ this.tables = tables;
this.pendingSplits = new HashMap<>();
this.shouldEnumerate = sourceState == null;
if (sourceState != null) {
@@ -133,13 +144,16 @@ public class AmazonDynamoDBSourceSplitEnumerator
private Set<AmazonDynamoDBSourceSplit> discoverySplits() {
Set<AmazonDynamoDBSourceSplit> allSplit = new HashSet<>();
- int totalSegments = amazonDynamoDBConfig.parallelScanThreads;
- int itemLimit = amazonDynamoDBConfig.scanItemLimit;
- for (int i = 0; i < totalSegments; i++) {
- AmazonDynamoDBSourceSplit split =
- new AmazonDynamoDBSourceSplit(i, totalSegments, itemLimit);
-
- allSplit.add(split);
+ for (AmazonDynamoDBSourceTable table : tables) {
+ int totalSegments = table.getConfig().parallelScanThreads;
+ int itemLimit = table.getConfig().scanItemLimit;
+ for (int i = 0; i < totalSegments; i++) {
+ AmazonDynamoDBSourceSplit split =
+ new AmazonDynamoDBSourceSplit(
+ i, totalSegments, itemLimit,
table.getTableId());
+
+ allSplit.add(split);
+ }
}
return allSplit;
}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceTable.java
similarity index 61%
copy from
seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
copy to
seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceTable.java
index 825cc3666d..af50b95691 100644
---
a/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceSplit.java
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceTable.java
@@ -17,24 +17,25 @@
package org.apache.seatunnel.connectors.seatunnel.amazondynamodb.source;
-import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
import lombok.AllArgsConstructor;
import lombok.Getter;
-import lombok.Setter;
-@AllArgsConstructor
+import java.io.Serializable;
+
+/** A DynamoDB table read by the source, with the read options and row type
used for it. */
@Getter
-@Setter
-public class AmazonDynamoDBSourceSplit implements SourceSplit {
-
- private static final long serialVersionUID = -5148142613656330674L;
- private Integer splitId;
- private Integer totalSegments;
- private Integer itemCount;
-
- @Override
- public String splitId() {
- return splitId.toString();
- }
+@AllArgsConstructor
+class AmazonDynamoDBSourceTable implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Identity carried by splits and rows; null for a single-table source. */
+ private final String tableId;
+
+ private final AmazonDynamoDBConfig config;
+
+ private final SeaTunnelRowType rowType;
}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBMultiTableSourceTest.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBMultiTableSourceTest.java
new file mode 100644
index 0000000000..30cbaf2d01
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBMultiTableSourceTest.java
@@ -0,0 +1,542 @@
+/*
+ * 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.amazondynamodb.source;
+
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import org.apache.seatunnel.api.common.metrics.AbstractMetricsContext;
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+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.event.EventListener;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
+import software.amazon.awssdk.services.dynamodb.model.ScanResponse;
+import software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+class AmazonDynamoDBMultiTableSourceTest {
+
+ private static final String CONNECTION =
+ "url = \"http://localhost:8000\"\n";
+ + "region = \"us-east-1\"\n"
+ + "access_key_id = \"dummy-key\"\n"
+ + "secret_access_key = \"dummy-secret\"\n";
+
+ private static final String ORDERS =
+ "{ table = orders, parallel_scan_threads = 2, scan_item_limit =
10,"
+ + " schema { fields { id = string, amount = int } } }";
+
+ private static final String CUSTOMERS =
+ "{ table = customers,"
+ + " schema { table = \"crm.customer_view\", fields { id =
string, name = string, vip = boolean } } }";
+
+ private static final String SINGLE_TABLE =
+ "table = orders\nschema { fields { id = string, amount = int }
}\n";
+
+ /**
+ * {@link AmazonDynamoDBSourceState} serialized by the single-table source
before splits carried
+ * a table identity: segments 0 and 1 of 2 with a scan limit of 5, pending
for readers 0 and 1.
+ */
+ private static final String LEGACY_STATE =
+
"rO0ABXNyAFlvcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0b3JzLnNlYXR1bm5lbC5hbWF6b25keW5h"
+ +
"bW9kYi5zb3VyY2UuQW1hem9uRHluYW1vREJTb3VyY2VTdGF0ZYhyTqknexWFAgADSQALYXNzaWduQ291"
+ +
"bnRaAA9zaG91bGRFbnVtZXJhdGVMAA1wZW5kaW5nU3BsaXRzdAAPTGphdmEvdXRpbC9NYXA7eHAAAAAC"
+ +
"AHNyABFqYXZhLnV0aWwuSGFzaE1hcAUH2sHDFmDRAwACRgAKbG9hZEZhY3RvckkACXRocmVzaG9sZHhw"
+ +
"P0AAAAAAAAx3CAAAABAAAAACc3IAEWphdmEubGFuZy5JbnRlZ2VyEuKgpPeBhzgCAAFJAAV2YWx1ZXhy"
+ +
"ABBqYXZhLmxhbmcuTnVtYmVyhqyVHQuU4IsCAAB4cAAAAABzcgATamF2YS51dGlsLkFycmF5TGlzdHiB"
+ +
"0h2Zx2GdAwABSQAEc2l6ZXhwAAAAAXcEAAAAAXNyAFlvcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0"
+ +
"b3JzLnNlYXR1bm5lbC5hbWF6b25keW5hbW9kYi5zb3VyY2UuQW1hem9uRHluYW1vREJTb3VyY2VTcGxp"
+ +
"dLiOH5Gj9p5OAgADTAAJaXRlbUNvdW50dAATTGphdmEvbGFuZy9JbnRlZ2VyO0wAB3NwbGl0SWRxAH4A"
+ +
"C0wADXRvdGFsU2VnbWVudHNxAH4AC3hwc3EAfgAFAAAABXEAfgAHc3EAfgAFAAAAAnhzcQB+AAUAAAAB"
+ + "c3EAfgAIAAAAAXcEAAAAAXNxAH4ACnEAfgANcQB+AA9xAH4ADnh4";
+
+ @Test
+ void singleTableConfigKeepsOneTableAndLegacySplits() throws Exception {
+ AmazonDynamoDBSource source = source(SINGLE_TABLE +
"parallel_scan_threads = 3\n");
+
+ Assertions.assertEquals(1, source.getProducedCatalogTables().size());
+ List<AmazonDynamoDBSourceSplit> splits = enumerate(source);
+ Assertions.assertEquals(3, splits.size());
+ for (AmazonDynamoDBSourceSplit split : splits) {
+ Assertions.assertNull(split.getTableId());
+ Assertions.assertEquals(split.getSplitId().toString(),
split.splitId());
+ Assertions.assertEquals(3, split.getTotalSegments());
+ Assertions.assertEquals(1, split.getItemCount());
+ }
+ }
+
+ @Test
+ void producesOneCatalogTablePerConfiguredTable() {
+ AmazonDynamoDBSource source = source("tables_configs = [" + ORDERS +
"," + CUSTOMERS + "]");
+
+ List<CatalogTable> tables = source.getProducedCatalogTables();
+ Assertions.assertEquals(
+ Arrays.asList("orders", "crm.customer_view"),
+ tables.stream()
+ .map(table ->
table.getTableId().toTablePath().toString())
+ .collect(Collectors.toList()));
+ Assertions.assertArrayEquals(
+ new String[] {"id", "amount"},
tables.get(0).getSeaTunnelRowType().getFieldNames());
+ Assertions.assertArrayEquals(
+ new String[] {"id", "name", "vip"},
+ tables.get(1).getSeaTunnelRowType().getFieldNames());
+ }
+
+ @Test
+ void enumeratesParallelScanSegmentsPerTable() throws Exception {
+ AmazonDynamoDBSource source =
+ source(
+ "parallel_scan_threads = 3\nscan_item_limit = 7\n"
+ + "tables_configs = ["
+ + ORDERS
+ + ","
+ + CUSTOMERS
+ + "]");
+
+ List<AmazonDynamoDBSourceSplit> splits = enumerate(source);
+
+ Map<String, List<AmazonDynamoDBSourceSplit>> byTable =
+ splits.stream()
+
.collect(Collectors.groupingBy(AmazonDynamoDBSourceSplit::getTableId));
+ Assertions.assertEquals(2, byTable.get("orders").size());
+ Assertions.assertEquals(3, byTable.get("crm.customer_view").size());
+ for (AmazonDynamoDBSourceSplit split : byTable.get("orders")) {
+ Assertions.assertEquals(2, split.getTotalSegments());
+ Assertions.assertEquals(10, split.getItemCount());
+ }
+ for (AmazonDynamoDBSourceSplit split :
byTable.get("crm.customer_view")) {
+ Assertions.assertEquals(3, split.getTotalSegments());
+ Assertions.assertEquals(7, split.getItemCount());
+ }
+ Assertions.assertEquals(
+ Arrays.asList(
+ "crm.customer_view:0",
+ "crm.customer_view:1",
+ "crm.customer_view:2",
+ "orders:0",
+ "orders:1"),
+ splits.stream()
+ .map(AmazonDynamoDBSourceSplit::splitId)
+ .sorted()
+ .collect(Collectors.toList()));
+ }
+
+ @Test
+ void readerRoutesRowsToTheirTable() throws Exception {
+ AmazonDynamoDBSource source = source("tables_configs = [" + ORDERS +
"," + CUSTOMERS + "]");
+ List<AmazonDynamoDBSourceSplit> splits = roundTrip(enumerate(source));
+ Map<String, List<Map<String, AttributeValue>>> data = new HashMap<>();
+ data.put(
+ "orders",
+ Arrays.asList(
+ item("id", s("o1"), "amount", n("10")),
+ item("id", s("o2"), "amount", n("20"))));
+ data.put(
+ "customers",
+ Collections.singletonList(
+ item(
+ "id",
+ s("c1"),
+ "name",
+ s("Ada"),
+ "vip",
+ AttributeValue.builder().bool(true).build())));
+ DynamoDbClient client = client(data);
+
+ List<SeaTunnelRow> rows = read(source, client, splits);
+
+ Assertions.assertEquals(
+ Arrays.asList("[o1, 10]", "[o2, 20]"),
+ rows.stream()
+ .filter(row -> "orders".equals(row.getTableId()))
+ .map(row -> Arrays.toString(row.getFields()))
+ .sorted()
+ .collect(Collectors.toList()));
+ Assertions.assertEquals(
+ Collections.singletonList("[c1, Ada, true]"),
+ rows.stream()
+ .filter(row ->
"crm.customer_view".equals(row.getTableId()))
+ .map(row -> Arrays.toString(row.getFields()))
+ .collect(Collectors.toList()));
+ Assertions.assertEquals(
+ 0,
+ rows.stream()
+ .filter(
+ row ->
+ !"orders".equals(row.getTableId())
+ &&
!"crm.customer_view".equals(row.getTableId()))
+ .count());
+
+ ArgumentCaptor<ScanRequest> requests =
ArgumentCaptor.forClass(ScanRequest.class);
+ Mockito.verify(client,
Mockito.atLeastOnce()).scanPaginator(requests.capture());
+ Map<String, List<ScanRequest>> requestsByTable =
+ requests.getAllValues().stream()
+
.collect(Collectors.groupingBy(ScanRequest::tableName));
+ Assertions.assertEquals(2, requestsByTable.get("orders").size());
+ Assertions.assertEquals(2, requestsByTable.get("customers").size());
+ for (ScanRequest request : requestsByTable.get("orders")) {
+ Assertions.assertEquals(2, request.totalSegments());
+ Assertions.assertEquals(10, request.limit());
+ }
+ }
+
+ @Test
+ void restoresStateWrittenBeforeSplitsCarriedTableIdentity() throws
Exception {
+ AmazonDynamoDBSourceState state =
deserialize(Base64.getDecoder().decode(LEGACY_STATE));
+ List<AmazonDynamoDBSourceSplit> legacySplits = new ArrayList<>();
+ state.getPendingSplits().values().forEach(legacySplits::addAll);
+ Assertions.assertEquals(2, legacySplits.size());
+ for (AmazonDynamoDBSourceSplit split : legacySplits) {
+ Assertions.assertNull(split.getTableId());
+ Assertions.assertEquals(split.getSplitId().toString(),
split.splitId());
+ }
+
+ AmazonDynamoDBSource source = source(SINGLE_TABLE);
+ TestingContext context = new TestingContext();
+ SourceSplitEnumerator<AmazonDynamoDBSourceSplit,
AmazonDynamoDBSourceState> enumerator =
+ source.restoreEnumerator(context, state);
+ enumerator.open();
+ enumerator.run();
+ enumerator.registerReader(0);
+ enumerator.registerReader(1);
+ Assertions.assertEquals(1, context.assigned.get(0).size());
+ Assertions.assertEquals(1, context.assigned.get(1).size());
+ Assertions.assertEquals(2,
enumerator.snapshotState(1L).getAssignCount());
+
+ DynamoDbClient client =
+ client(
+ Collections.singletonMap(
+ "orders",
+ Collections.singletonList(item("id", s("o1"),
"amount", n("10")))));
+ AmazonDynamoDBSourceSplit firstSegment =
+ legacySplits.stream().filter(split -> split.getSplitId() ==
0).findFirst().get();
+ List<SeaTunnelRow> rows = read(source, client,
Collections.singletonList(firstSegment));
+ Assertions.assertEquals(1, rows.size());
+ // single-table rows keep the default table id, as before
+ Assertions.assertEquals(new SeaTunnelRow(0).getTableId(),
rows.get(0).getTableId());
+ Assertions.assertEquals("[o1, 10]",
Arrays.toString(rows.get(0).getFields()));
+ ArgumentCaptor<ScanRequest> request =
ArgumentCaptor.forClass(ScanRequest.class);
+ Mockito.verify(client,
Mockito.atLeastOnce()).scanPaginator(request.capture());
+ Assertions.assertEquals("orders", request.getValue().tableName());
+ Assertions.assertEquals(5, request.getValue().limit());
+ Assertions.assertEquals(2, request.getValue().totalSegments());
+ }
+
+ @Test
+ void readerRejectsSplitsThatDoNotMatchTheConfiguredTables() throws
Exception {
+ AmazonDynamoDBSource multiTable =
+ source("tables_configs = [" + ORDERS + "," + CUSTOMERS + "]");
+ DynamoDbClient client = client(Collections.emptyMap());
+
+ SeaTunnelRuntimeException legacySplit =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+ read(
+ multiTable,
+ client,
+ Collections.singletonList(
+ new
AmazonDynamoDBSourceSplit(0, 1, 1))));
+ Assertions.assertTrue(
+ legacySplit.getMessage().contains("Cannot restore a
single-table"),
+ legacySplit.getMessage());
+
+ SeaTunnelRuntimeException unknownTable =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
+ read(
+ source(SINGLE_TABLE),
+ client,
+ Collections.singletonList(
+ new
AmazonDynamoDBSourceSplit(0, 1, 1, "orders"))));
+ Assertions.assertTrue(
+ unknownTable.getMessage().contains("Unknown table identity"),
+ unknownTable.getMessage());
+ }
+
+ @Test
+ void multiTableSplitsSurviveSerialization() throws Exception {
+ AmazonDynamoDBSourceSplit split = roundTrip(new
AmazonDynamoDBSourceSplit(1, 2, 3, "t"));
+ Assertions.assertEquals("t", split.getTableId());
+ Assertions.assertEquals("t:1", split.splitId());
+ }
+
+ @Test
+ void optionRuleAcceptsSingleTableOrTablesConfigs() {
+ validate(SINGLE_TABLE);
+ validate("tables_configs = [" + ORDERS + "," + CUSTOMERS + "]");
+ }
+
+ @Test
+ void acceptsUnquotedNumericTableName() {
+ AmazonDynamoDBSource source =
+ source("tables_configs = [{ table = 2024, schema { fields { id
= string } } }]");
+ Assertions.assertEquals(
+ "2024",
+
source.getProducedCatalogTables().get(0).getTableId().toTablePath().toString());
+ }
+
+ @Test
+ void optionRuleRejectsInvalidTableSelection() {
+ assertInvalid("", "exactly one option must be set");
+ assertInvalid(
+ SINGLE_TABLE + "tables_configs = [" + ORDERS + "]",
+ "mutually exclusive, but multiple are set");
+ assertInvalid("table = orders\n", "'schema' must be configured when
using a root-level");
+ assertInvalid("tables_configs = []\n", "is not empty");
+ assertInvalid(
+ "schema { fields { id = string } }\ntables_configs = [" +
ORDERS + "]",
+ "root-level 'schema' cannot be used with 'tables_configs'");
+ }
+
+ @Test
+ void optionRuleRejectsInvalidTableEntries() {
+ assertInvalid(
+ "tables_configs = [{ schema { fields { id = string } } }]",
+ "tables_configs[0]: 'table' must be configured and non-blank");
+ assertInvalid(
+ "tables_configs = ["
+ + ORDERS
+ + ", { table = \" \", schema { fields { id = string }
} }]",
+ "tables_configs[1]: 'table' must be configured and non-blank");
+ assertInvalid(
+ "tables_configs = [{ table = orders }]",
+ "tables_configs[0]: 'schema' must be configured and
non-empty");
+ assertInvalid(
+ "tables_configs = [" + ORDERS + "," + ORDERS + "]",
+ "tables_configs[1]: duplicate table identity 'orders'");
+ assertInvalid(
+ "tables_configs = ["
+ + ORDERS
+ + ", { table = orders_archive, schema { table =
orders, fields { id = string } } }]",
+ "tables_configs[1]: duplicate table identity 'orders'");
+ assertInvalid(
+ "tables_configs = [{ table = orders, region = \"eu-west-1\",
schema { fields { id = string } } }]",
+ "tables_configs[0]: 'region' must be configured at source
level");
+ assertInvalid(
+ "tables_configs = [{ table = orders, URL =
\"http://other:8000\";, schema { fields { id = string } } }]",
+ "tables_configs[0]: unsupported table option 'URL'");
+ assertInvalid(
+ "tables_configs = [{ table = orders, batch_size = 5, schema {
fields { id = string } } }]",
+ "tables_configs[0]: unsupported table option 'batch_size'");
+ assertInvalid(
+ "tables_configs = [{ table = orders, parallel_scan_threads =
0, schema { fields { id = string } } }]",
+ "tables_configs[0]: 'scan_item_limit' and
'parallel_scan_threads' must be positive");
+ assertInvalid(
+ "tables_configs = [{ table = orders, scan_item_limit = many,
schema { fields { id = string } } }]",
+ "tables_configs[0]: invalid table entry");
+ }
+
+ private static void validate(String tableConfig) {
+ ConfigValidator.of(config(tableConfig))
+ .validate(new AmazonDynamoDBSourceFactory().optionRule());
+ }
+
+ private static void assertInvalid(String tableConfig, String
expectedMessage) {
+ OptionValidationException exception =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validate(tableConfig));
+ Assertions.assertTrue(
+ exception.getMessage().contains(expectedMessage),
exception.getMessage());
+
Assertions.assertFalse(exception.getMessage().contains("dummy-secret"));
+ }
+
+ private static ReadonlyConfig config(String tableConfig) {
+ return ReadonlyConfig.fromConfig(ConfigFactory.parseString(CONNECTION
+ tableConfig));
+ }
+
+ private static AmazonDynamoDBSource source(String tableConfig) {
+ ReadonlyConfig config = config(tableConfig);
+ AmazonDynamoDBSourceFactory factory = new
AmazonDynamoDBSourceFactory();
+ ConfigValidator.of(config).validate(factory.optionRule());
+ return (AmazonDynamoDBSource)
+ factory.<SeaTunnelRow, AmazonDynamoDBSourceSplit,
AmazonDynamoDBSourceState>
+ createSource(
+ new TableSourceFactoryContext(
+ config,
+
AmazonDynamoDBMultiTableSourceTest.class
+ .getClassLoader()))
+ .createSource();
+ }
+
+ private static List<AmazonDynamoDBSourceSplit>
enumerate(AmazonDynamoDBSource source)
+ throws Exception {
+ TestingContext context = new TestingContext();
+ SourceSplitEnumerator<AmazonDynamoDBSourceSplit,
AmazonDynamoDBSourceState> enumerator =
+ source.createEnumerator(context);
+ enumerator.open();
+ enumerator.run();
+ List<AmazonDynamoDBSourceSplit> splits = new ArrayList<>();
+ context.assigned.values().forEach(splits::addAll);
+ return splits;
+ }
+
+ private static List<SeaTunnelRow> read(
+ AmazonDynamoDBSource source,
+ DynamoDbClient client,
+ List<AmazonDynamoDBSourceSplit> splits)
+ throws Exception {
+ SourceReader.Context context =
Mockito.mock(SourceReader.Context.class);
+ Mockito.when(context.getBoundedness()).thenReturn(Boundedness.BOUNDED);
+ AmazonDynamoDBSourceReader reader =
+ (AmazonDynamoDBSourceReader) source.createReader(context);
+ reader.dynamoDbClient = client;
+ reader.addSplits(splits);
+ reader.handleNoMoreSplits();
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ Collector<SeaTunnelRow> collector =
+ new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow record) {
+ rows.add(record);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ };
+ for (int i = 0; i < splits.size(); i++) {
+ reader.pollNext(collector);
+ }
+ return rows;
+ }
+
+ /** Returns each table's items from segment 0 and nothing from the other
segments. */
+ private static DynamoDbClient client(Map<String, List<Map<String,
AttributeValue>>> data) {
+ DynamoDbClient client = Mockito.mock(DynamoDbClient.class);
+ Mockito.when(client.scanPaginator(Mockito.any(ScanRequest.class)))
+ .thenAnswer(
+ invocation ->
+ new ScanIterable(
+ client, invocation.getArgument(0,
ScanRequest.class)));
+ Mockito.when(client.scan(Mockito.any(ScanRequest.class)))
+ .thenAnswer(
+ invocation -> {
+ ScanRequest request = invocation.getArgument(0,
ScanRequest.class);
+ List<Map<String, AttributeValue>> items =
+ request.segment() == 0
+ ? data.getOrDefault(
+ request.tableName(),
Collections.emptyList())
+ : Collections.emptyList();
+ return ScanResponse.builder().items(items).build();
+ });
+ return client;
+ }
+
+ private static Map<String, AttributeValue> item(Object... keyValues) {
+ Map<String, AttributeValue> item = new HashMap<>();
+ for (int i = 0; i < keyValues.length; i += 2) {
+ item.put((String) keyValues[i], (AttributeValue) keyValues[i + 1]);
+ }
+ return item;
+ }
+
+ private static AttributeValue s(String value) {
+ return AttributeValue.builder().s(value).build();
+ }
+
+ private static AttributeValue n(String value) {
+ return AttributeValue.builder().n(value).build();
+ }
+
+ private static <T> T roundTrip(T value) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
+ output.writeObject(value);
+ }
+ return deserialize(bytes.toByteArray());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T deserialize(byte[] bytes) throws Exception {
+ try (ObjectInputStream input = new ObjectInputStream(new
ByteArrayInputStream(bytes))) {
+ return (T) input.readObject();
+ }
+ }
+
+ private static class TestingContext
+ implements
SourceSplitEnumerator.Context<AmazonDynamoDBSourceSplit> {
+ private final Map<Integer, List<AmazonDynamoDBSourceSplit>> assigned =
new HashMap<>();
+ private final MetricsContext metricsContext = new
AbstractMetricsContext() {};
+
+ @Override
+ public int currentParallelism() {
+ return 2;
+ }
+
+ @Override
+ public Set<Integer> registeredReaders() {
+ return new HashSet<>(Arrays.asList(0, 1));
+ }
+
+ @Override
+ public void assignSplit(int subtaskId, List<AmazonDynamoDBSourceSplit>
splits) {
+ assigned.computeIfAbsent(subtaskId, ignored -> new
ArrayList<>()).addAll(splits);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {}
+
+ @Override
+ public void sendEventToSourceReader(int subtaskId, SourceEvent event)
{}
+
+ @Override
+ public MetricsContext getMetricsContext() {
+ return metricsContext;
+ }
+
+ @Override
+ public EventListener getEventListener() {
+ return event -> {};
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReaderTest.java
b/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReaderTest.java
new file mode 100644
index 0000000000..9cf7aa3853
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-amazondynamodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceReaderTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.amazondynamodb.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.mockito.Mockito;
+
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
+import software.amazon.awssdk.services.dynamodb.model.ScanResponse;
+import software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+class AmazonDynamoDBSourceReaderTest {
+
+ @Test
+ @Timeout(30)
+ void readsSegmentOnceBeforeNoMoreSplitsArrives() throws Exception {
+ AmazonDynamoDBSourceReader reader = reader();
+ AtomicInteger scans = new AtomicInteger();
+ DynamoDbClient client = Mockito.mock(DynamoDbClient.class);
+ Mockito.when(client.scan(Mockito.any(ScanRequest.class)))
+ .thenAnswer(
+ invocation -> {
+ // The enumerator's no-more-splits signal arrives
late.
+ if (scans.incrementAndGet() >= 3) {
+ reader.handleNoMoreSplits();
+ }
+ return ScanResponse.builder()
+ .items(Arrays.asList(item("o1"),
item("o2")))
+ .build();
+ });
+
+ List<SeaTunnelRow> rows = read(reader, client);
+
+ Assertions.assertEquals(2, rows.size());
+ Assertions.assertEquals(1, scans.get());
+ }
+
+ @Test
+ @Timeout(30)
+ void readsEveryPageOfSegment() throws Exception {
+ AmazonDynamoDBSourceReader reader = reader();
+ List<ScanRequest> requests = new ArrayList<>();
+ DynamoDbClient client = Mockito.mock(DynamoDbClient.class);
+ Mockito.when(client.scan(Mockito.any(ScanRequest.class)))
+ .thenAnswer(
+ invocation -> {
+ ScanRequest request = invocation.getArgument(0,
ScanRequest.class);
+ requests.add(request);
+ if (!request.hasExclusiveStartKey()) {
+ return ScanResponse.builder()
+ .items(Arrays.asList(item("o1"),
item("o2")))
+ .lastEvaluatedKey(item("o2"))
+ .build();
+ }
+ return ScanResponse.builder()
+
.items(Collections.singletonList(item("o3")))
+ .build();
+ });
+
+ List<SeaTunnelRow> rows = read(reader, client);
+
+ Assertions.assertEquals(
+ Arrays.asList("o1", "o2", "o3"),
+ rows.stream().map(row ->
row.getField(0)).collect(Collectors.toList()));
+ Assertions.assertEquals(2, requests.size());
+ Assertions.assertEquals(item("o2"),
requests.get(1).exclusiveStartKey());
+ }
+
+ private static AmazonDynamoDBSourceReader reader() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("url", "http://localhost:8000";);
+ options.put("region", "us-east-1");
+ options.put("access_key_id", "id");
+ options.put("secret_access_key", "key");
+ options.put("table", "orders");
+ AmazonDynamoDBConfig config = new
AmazonDynamoDBConfig(ReadonlyConfig.fromMap(options));
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id"}, new SeaTunnelDataType<?>[]
{BasicType.STRING_TYPE});
+ return new AmazonDynamoDBSourceReader(
+ Mockito.mock(SourceReader.Context.class), config, rowType);
+ }
+
+ private static List<SeaTunnelRow> read(AmazonDynamoDBSourceReader reader,
DynamoDbClient client)
+ throws Exception {
+ Mockito.when(client.scanPaginator(Mockito.any(ScanRequest.class)))
+ .thenAnswer(
+ invocation ->
+ new ScanIterable(
+ client, invocation.getArgument(0,
ScanRequest.class)));
+ reader.dynamoDbClient = client;
+ reader.addSplits(Collections.singletonList(new
AmazonDynamoDBSourceSplit(0, 1, 10)));
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ reader.pollNext(
+ new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow record) {
+ rows.add(record);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ });
+ return rows;
+ }
+
+ private static Map<String, AttributeValue> item(String id) {
+ return Collections.singletonMap("id",
AttributeValue.builder().s(id).build());
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/pom.xml
index ff0bcbe41c..dc14bc8280 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/pom.xml
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/pom.xml
@@ -36,6 +36,12 @@
</dependencies>
</dependencyManagement>
<dependencies>
+ <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-amazondynamodb</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/amazondynamodb/AmazondynamodbIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/amazondynamodb/AmazondynamodbIT.java
index d6b09fb7bc..da4448410e 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/amazondynamodb/AmazondynamodbIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/amazondynamodb/AmazondynamodbIT.java
@@ -84,8 +84,11 @@ public class AmazondynamodbIT extends TestSuiteBase
implements TestResource {
private static final String AMAZONDYNAMODB_CONTAINER_HOST =
"dynamodb-host";
private static final int AMAZONDYNAMODB_CONTAINER_PORT = 8000;
private static final String AMAZONDYNAMODB_JOB_CONFIG =
"/amazondynamodbIT_source_to_sink.conf";
+ private static final String AMAZONDYNAMODB_MULTI_TABLE_JOB_CONFIG =
+ "/amazondynamodbIT_multi_table_source.conf";
private static final String SINK_TABLE = "sink_table";
private static final String SOURCE_TABLE = "source_table";
+ private static final String CUSTOMER_TABLE = "customer_table";
private static final String PARTITION_KEY = "id";
private GenericContainer<?> dynamoDB;
@@ -102,6 +105,13 @@ public class AmazondynamodbIT extends TestSuiteBase
implements TestResource {
clearSinkTable();
}
+ @TestTemplate
+ public void testMultiTableSource(TestContainer container) throws Exception
{
+ Container.ExecResult execResult =
+ container.executeJob(AMAZONDYNAMODB_MULTI_TABLE_JOB_CONFIG);
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
@BeforeAll
@Override
public void startUp() throws Exception {
@@ -153,11 +163,20 @@ public class AmazondynamodbIT extends TestSuiteBase
implements TestResource {
createTable(dynamoDbClient, SOURCE_TABLE);
createTable(dynamoDbClient, SINK_TABLE);
+ createTable(dynamoDbClient, CUSTOMER_TABLE);
}
private void batchInsertData() {
dynamoDbClient.putItem(
PutItemRequest.builder().tableName(SOURCE_TABLE).item(randomRow()).build());
+ for (int i = 1; i <= 3; i++) {
+ Map<String, AttributeValue> customer = new HashMap<>();
+ customer.put(PARTITION_KEY, AttributeValue.builder().s("c" +
i).build());
+ customer.put("name", AttributeValue.builder().s("customer-" +
i).build());
+ customer.put("vip", AttributeValue.builder().bool(i % 2 ==
0).build());
+ dynamoDbClient.putItem(
+
PutItemRequest.builder().tableName(CUSTOMER_TABLE).item(customer).build());
+ }
}
private void clearSinkTable() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/resources/amazondynamodbIT_multi_table_source.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/resources/amazondynamodbIT_multi_table_source.conf
new file mode 100644
index 0000000000..cc6e679ca3
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-amazondynamodb-e2e/src/test/resources/amazondynamodbIT_multi_table_source.conf
@@ -0,0 +1,104 @@
+#
+# 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 = 1
+ job.mode = "BATCH"
+}
+
+source {
+ AmazonDynamoDB {
+ url = "http://dynamodb-host:8000";
+ region = "us-east-1"
+ access_key_id = "dummy-key"
+ secret_access_key = "dummy-secret"
+ scan_item_limit = 2
+ parallel_scan_threads = 2
+ tables_configs = [
+ {
+ table = "source_table"
+ parallel_scan_threads = 3
+ schema = {
+ fields {
+ id = string
+ c_string = string
+ c_int = int
+ }
+ }
+ },
+ {
+ table = "customer_table"
+ schema = {
+ table = "crm.customers"
+ fields {
+ id = string
+ name = string
+ vip = boolean
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules {
+ table-names = ["source_table", "crm.customers"]
+ tables_configs = [
+ {
+ table_path = "source_table"
+ row_rules = [
+ { rule_type = MAX_ROW, rule_value = 1 },
+ { rule_type = MIN_ROW, rule_value = 1 }
+ ]
+ field_rules = [
+ {
+ field_name = c_string
+ field_type = string
+ field_value = [{ equals_to = "string" }]
+ },
+ {
+ field_name = c_int
+ field_type = int
+ field_value = [{ equals_to = 1 }]
+ }
+ ]
+ },
+ {
+ table_path = "crm.customers"
+ row_rules = [
+ { rule_type = MAX_ROW, rule_value = 3 },
+ { rule_type = MIN_ROW, rule_value = 3 }
+ ]
+ field_rules = [
+ {
+ field_name = name
+ field_type = string
+ field_value = [{ rule_type = NOT_NULL }]
+ },
+ {
+ field_name = vip
+ field_type = boolean
+ field_value = [{ rule_type = NOT_NULL }]
+ }
+ ]
+ }
+ ]
+ }
+ }
+}
