This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 c5ea35b929 [Docs] Align multi-table writer and replica-selection docs
with MultiTableSinkWriter (#11925)
c5ea35b929 is described below
commit c5ea35b92942ef8117abb497d5961fcacc196540
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Sun Aug 23 19:00:46 2026 +0530
[Docs] Align multi-table writer and replica-selection docs with
MultiTableSinkWriter (#11925)
---
docs/en/architecture/features/multi-table.md | 208 ++++++++++++++++-----------
docs/zh/architecture/features/multi-table.md | 43 ++++--
2 files changed, 156 insertions(+), 95 deletions(-)
diff --git a/docs/en/architecture/features/multi-table.md
b/docs/en/architecture/features/multi-table.md
index 43d115a832..339b54b933 100644
--- a/docs/en/architecture/features/multi-table.md
+++ b/docs/en/architecture/features/multi-table.md
@@ -242,85 +242,88 @@ public class MultiTableSink<IN, StateT, CommitInfoT,
AggregatedCommitInfoT>
### 4.2 Writer: Multi-Table Writing with Replicas
-```java
-public class MultiTableSinkWriter<IN, CommitInfoT, StateT>
- implements SinkWriter<IN, CommitInfoT, StateT> {
-
- // Writers per table (multiple replicas per table)
- private final Map<SinkIdentifier, SinkWriter<IN, CommitInfoT, StateT>>
writers;
-
- // Replica count per table
- private final int replicaNum;
-
- // Context
- private final int writerIndex; // This writer's global index
-
- @Override
- public void write(IN element) throws IOException {
- SeaTunnelRow row = (SeaTunnelRow) element;
-
- // 1. Determine target table
- TablePath tablePath = row.getTablePath();
-
- // 2. Select replica for this table (load balancing)
- int replicaIndex = selectReplica(tablePath, row);
-
- // 3. Get writer for (table, replica)
- SinkIdentifier identifier = new SinkIdentifier(
- new TableIdentifier(tablePath),
- replicaIndex
- );
+`MultiTableSinkWriter` does not write rows inline. Each row is routed to one of
+`blockingQueues`, and a `MultiTableWriterRunnable` worker drains that queue
and writes
+through the sub-writers assigned to it. The queue index *is* the replica index.
- SinkWriter<IN, CommitInfoT, StateT> writer = writers.get(identifier);
+The listing below keeps the real class, field, and method names so it can be
read
+side by side with
+`seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java`.
+It is **simplified**: schema-change short-circuiting, quarantine checks,
retry, and
+exception wrapping are elided. It is not a copy of the source.
- // 4. Write to selected writer
- writer.write(element);
- }
+```java
+public class MultiTableSinkWriter
+ implements SinkWriter<SeaTunnelRow, MultiTableCommitInfo,
MultiTableState>,
+ SupportSchemaEvolutionSinkWriter {
- private int selectReplica(TablePath tablePath, SeaTunnelRow row) {
- // If primary key is available, route stably by primary key hash.
- Optional<Object> primaryKey = extractPrimaryKeyIfPresent(row);
- if (primaryKey.isPresent()) {
- return Math.abs(primaryKey.get().hashCode()) % replicaNum;
- }
+ // Sub-writers, keyed by (table identifier, replica index)
+ private final Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>>
sinkWriters;
- // Otherwise, distribute across replicas (no stable routing guarantee).
- return (int) (System.nanoTime() % replicaNum);
- }
+ // Primary-key field index per table id. An empty Optional means "table
has no
+ // primary key"; a missing key means "no sub-writer is registered for this
table".
+ private final ConcurrentMap<String, Optional<Integer>> sinkPrimaryKeys;
- @Override
- public Optional<CommitInfoT> prepareCommit(long checkpointId) throws
IOException {
- // Collect commit info from all writers
- List<CommitInfoT> allCommitInfos = new ArrayList<>();
+ // One queue per writer thread. The queue index is the replica index.
+ private final List<BlockingQueue<MultiTableWriterRunnable.QueueElement>>
blockingQueues;
- for (SinkWriter<IN, CommitInfoT, StateT> writer : writers.values()) {
- Optional<CommitInfoT> commitInfo =
writer.prepareCommit(checkpointId);
- commitInfo.ifPresent(allCommitInfos::add);
- }
+ // Sub-writers grouped by queue index; each runnable owns exactly one group
+ private final List<ConcurrentMap<SinkIdentifier, SinkWriter<SeaTunnelRow,
?, ?>>>
+ sinkWritersWithIndex;
- // Wrap in multi-table commit info
- return Optional.of((CommitInfoT) new
MultiTableCommitInfo(allCommitInfos));
- }
+ private final List<MultiTableWriterRunnable> runnable;
+ private final Random random = new Random();
+ private final MultiTableFailurePolicy failurePolicy;
@Override
- public List<StateT> snapshotState(long checkpointId) throws IOException {
- // Snapshot all writers
- List<StateT> allStates = new ArrayList<>();
-
- for (Map.Entry<SinkIdentifier, SinkWriter> entry : writers.entrySet())
{
- List<StateT> states = entry.getValue().snapshotState(checkpointId);
-
- // Tag states with sink identifier for recovery
- for (StateT state : states) {
- allStates.add(wrapWithIdentifier(entry.getKey(), state));
+ public void write(SeaTunnelRow element) throws IOException {
+ ensureQueueWorkersSubmitted();
+ subSinkErrorCheck();
+
+ // 1. Rows carry the table id, not a TablePath
+ String tableId = element.getTableId();
+ Optional<Integer> primaryKey = tableId == null ? null :
sinkPrimaryKeys.get(tableId);
+
+ if ((primaryKey == null && sinkPrimaryKeys.size() == 1)
+ || (primaryKey != null && !primaryKey.isPresent())) {
+ // 2a. No primary key to route on: spread rows across queues.
+ // No per-key ordering guarantee.
+ int index = random.nextInt(blockingQueues.size());
+ offerRowElement(index, element);
+
+ } else if (primaryKey == null) {
+ // 2b. No sub-writer registered for this table. Quarantine it and
keep the
+ // other tables running, or fail the job, per the failure
policy.
+ if (failurePolicy.continueOtherTables()) {
+ handleTableFailure(tableId,
MultiTableFailurePhase.RUNTIME_WRITE, ...);
+ return;
+ }
+ throw new RuntimeException("multi table sink can not write table:
" + tableId);
+
+ } else {
+ // 2c. Primary key present: route by its hash so equal keys always
reach the
+ // same queue, which is what preserves per-key ordering.
+ Object object = element.getField(primaryKey.get());
+ int index = 0;
+ if (object != null) {
+ // Known issue: Math.abs(Integer.MIN_VALUE) returns
Integer.MIN_VALUE
+ // unchanged (still negative), so a key hashing to it yields a
negative
+ // queue index. See section 5.3.
+ index = Math.abs(object.hashCode()) % blockingQueues.size();
}
+ offerRowElement(index, element);
}
-
- return allStates;
}
}
```
+Because rows are queued rather than written inline, the checkpoint methods
must first
+drain the queues. Both `prepareCommit(long)` and `snapshotState(long)` begin
with
+`checkQueueRemain()`, then call the corresponding method on every sub-writer
while
+holding that sub-writer's `MultiTableWriterRunnable` lock. Their real bodies
are
+dominated by parallel task submission, per-table retry, and failure-policy
handling,
+and are not reproduced here.
+
### 4.3 Committer: Multi-Table Commit Coordination
```java
@@ -393,18 +396,37 @@ sink {
### 5.3 Replica Selection Strategies
-**Hash-Based (when primary key is available)**:
+Both strategies select a queue index in `[0, blockingQueues.size())`. See
+`MultiTableSinkWriter.write(SeaTunnelRow)` for the full branch structure.
+
+**Hash-based (primary key available)** — the same key always reaches the same
queue,
+which is what preserves ordering for that key:
+
```java
-// Ensures same primary key always goes to same replica (order preservation)
-int replica = Math.abs(primaryKey.hashCode()) % replicaNum;
+int index = Math.abs(object.hashCode()) % blockingQueues.size();
```
-**Random (when primary key is not available)**:
+:::caution Known issue
+
+`Math.abs(Integer.MIN_VALUE)` returns `Integer.MIN_VALUE`, which is still
negative, so a
+key whose hash is exactly `Integer.MIN_VALUE` produces a negative index
whenever the queue
+count is not a power of two, and the subsequent `blockingQueues.get(index)`
throws
+`IndexOutOfBoundsException`. This page documents the behaviour currently on
`dev`; the
+defect is tracked in
[#11720](https://github.com/apache/seatunnel/issues/11720), and this
+section should be updated when a fix lands.
+
+:::
+
+**Random (no primary key available)** — spreads load, with no stable routing
guarantee:
+
```java
-// Distributes load across replicas (no stable routing guarantee)
-int replica = (int) (System.nanoTime() % replicaNum);
+int index = random.nextInt(blockingQueues.size());
```
+`random.nextInt(bound)` is used rather than a time-derived expression such as
+`System.nanoTime() % n`. `System.nanoTime()` is documented as possibly
negative, so that
+expression can produce a negative index for exactly the same reason `Math.abs`
can above.
+
## 6. Schema Management in Multi-Table
### 6.1 Independent Schemas
@@ -424,21 +446,39 @@ public class MultiTableSink {
### 6.2 Schema Evolution Routing
+A schema change is not applied by looping over sub-writers directly. It is
enqueued as a
+**barrier** onto every queue, so each worker applies it at the same position
in its own
+row stream and no row crosses the change out of order. Simplified, with the
failure
+collection and post-enqueue error recheck elided:
+
```java
-public class MultiTableSinkWriter {
- public void handleSchemaChange(SchemaChangeEvent event) {
- // Route schema change to correct table writer
- TablePath tablePath = event.getTableId().toTablePath();
-
- // Apply to all replicas of this table
- for (int i = 0; i < replicaNum; i++) {
- SinkIdentifier identifier = new SinkIdentifier(
- new TableIdentifier(tablePath),
- i
- );
-
- SinkWriter writer = writers.get(identifier);
- writer.applySchemaChange(event);
+public class MultiTableSinkWriter implements SupportSchemaEvolutionSinkWriter {
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent event) throws IOException {
+ subSinkErrorCheck();
+
+ // Events for tables this writer does not serve return immediately,
+ // without waking the queue workers.
+ if (!hasSourceMatchedWriter(event)) {
+ return;
+ }
+
+ ensureQueueWorkersSubmitted();
+ subSinkErrorCheck();
+ enqueueSchemaChangeBarrier(event);
+ }
+
+ private void enqueueSchemaChangeBarrier(SchemaChangeEvent event) throws
IOException {
+ // One barrier shared by all workers; it releases once every worker
has arrived.
+ SchemaChangeBarrier barrier =
+ new SchemaChangeBarrier(
+ event,
+ runnable.size(),
+ e -> dispatchSchemaChangeToTargets(e,
schemaChangeFailures));
+
+ for (BlockingQueue<MultiTableWriterRunnable.QueueElement> queue :
blockingQueues) {
+ offerQueueElement(queue,
MultiTableWriterRunnable.schemaChangeRequest(barrier));
}
}
}
diff --git a/docs/zh/architecture/features/multi-table.md
b/docs/zh/architecture/features/multi-table.md
index cbc3299161..71067d5a80 100644
--- a/docs/zh/architecture/features/multi-table.md
+++ b/docs/zh/architecture/features/multi-table.md
@@ -127,21 +127,24 @@ MultiTableSink 是一个“按表路由 + 可多副本并行写入”的 Sink:
### 4.2 写入器: 带副本的多表写入
写入器的关键流程:
-1. 从输入记录中解析 TablePath(tableId)
-2. 为该表选择一个 writer 副本(replicaIndex)
-3. 路由到 (TablePath, replicaIndex) 对应的底层 writer 执行写入
+1. 从输入记录中读取 tableId（`SeaTunnelRow.getTableId()`）
+2. 依据该表登记的主键字段信息选择一个队列下标（即副本下标 replicaIndex）
+3. 将记录投递到对应的 `blockingQueues`，由 `MultiTableWriterRunnable` 工作线程消费并调用底层 writer
写入
+
+注意：记录并非在 `write()` 中同步写入底层 writer，而是先入队、再由工作线程异步写出。队列下标即副本下标。
副本选择需要兼顾两类诉求:
- **顺序性/一致落点**: 对同一主键（或唯一键）相关的记录尽量路由到同一副本，降低乱序与写入冲突风险
- **吞吐量**: 在不破坏顺序性要求的前提下，尽量分散写入压力
-在当前 MultiTableSinkWriter 的实现中，副本选择主要依据“主键信息是否可用”：
-- 有主键：对主键字段做哈希，稳定映射到某个副本
-- 无主键：使用随机策略在副本间分配
+在当前 MultiTableSinkWriter 的实现中，副本选择分为三个分支：
+- **有主键**：对主键字段值做哈希，稳定映射到某个队列
+- **无主键**：使用随机策略（`Random.nextInt`）在队列间分配
+- **该表未登记 writer**：按 `MultiTableFailurePolicy` 处理——或隔离该表并继续写入其他表，或直接抛出异常终止作业
这意味着“是否按 rowKind（INSERT/UPDATE/DELETE）切换策略”不是该实现的默认行为；如果需要按 rowKind 细分策略，应以
connector/实现代码为准。
-在 checkpoint 边界:
+在 checkpoint 边界（两者都先调用 `checkQueueRemain()` 排空队列）:
- prepareCommit: 汇总所有表/所有副本的 CommitInfo，并打包为多表级提交信息
- snapshotState: 快照所有 writer 状态；恢复时必须能通过 SinkIdentifier 将状态路由回正确的(表,副本)
@@ -185,17 +188,31 @@ sink {
### 5.3 副本选择策略
+两种策略都在 `[0, blockingQueues.size())` 范围内选择队列下标，完整分支结构参见
+`MultiTableSinkWriter.write(SeaTunnelRow)`。
+
**基于主键哈希（稳定路由）**:
要点:
- 以主键（或业务唯一键）做哈希，将同一键稳定映射到同一副本
-- 典型映射: $replica = hash(pk) \bmod replicaNum$
+- 当前实现: $replica = \mathrm{Math.abs}(hash(pk)) \bmod replicaNum$
+
+:::caution 已知问题
+
+`Math.abs(Integer.MIN_VALUE)` 仍返回 `Integer.MIN_VALUE`（负数），因此当主键哈希恰好为
+`Integer.MIN_VALUE` 且副本数不是 2 的幂时会得到负的下标，随后的
+`blockingQueues.get(index)` 会抛出 `IndexOutOfBoundsException`。本页描述的是 `dev`
+分支当前的实际行为；该缺陷记录在
+[#11720](https://github.com/apache/seatunnel/issues/11720)，修复合入后需同步更新本节。
+
+:::
**随机（无主键兜底）**:
要点:
- 当记录缺少主键字段信息时，无法提供稳定落点
-- 使用随机分配在副本间扩散压力，但不保证同一键的顺序性
+- 使用 `Random.nextInt(blockingQueues.size())` 在副本间扩散压力，但不保证同一键的顺序性
+- 不使用 `System.nanoTime() % replicaNum` 之类的写法：`nanoTime()` 可能为负，同样会产生负下标
## 6. 多表中的模式管理
@@ -210,8 +227,12 @@ sink {
模式演化需要被路由到“正确的表”，并应用到该表的所有 writer 副本:
1. 从 SchemaChangeEvent 中解析出 TablePath
-2. 选择该表对应的 schema/元数据更新逻辑
-3. 将变更广播到该表的所有副本 writer，保证后续写入使用一致的 schema
+2. 若当前 writer 不服务该表（`hasSourceMatchedWriter` 返回 false），直接返回，不唤醒队列工作线程
+3. 将变更以**屏障（`SchemaChangeBarrier`）**的形式投递到所有 `blockingQueues`
+4. 每个工作线程在各自行流的同一位置应用该变更，保证不会有记录跨越 schema 变更乱序写出
+
+注意：模式变更并非直接遍历各副本 writer 同步调用，而是与普通行记录共用同一条队列路径，
+以此保证变更与行记录之间的相对顺序。
## 7. 数据流示例
