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 5050fbaa34 [Improve][CLI] Add conditional-routing skill,
PostgreSQL-CDC prerequisites, and wide-DAG wiring rules (#11659)
5050fbaa34 is described below
commit 5050fbaa345b48a3ef4aa81c777ff8f8b3c69056
Author: SEZ <[email protected]>
AuthorDate: Tue Aug 11 23:26:49 2026 +0900
[Improve][CLI] Add conditional-routing skill, PostgreSQL-CDC prerequisites,
and wide-DAG wiring rules (#11659)
Co-authored-by: SEZ9 <[email protected]>
---
seatunnel-cli/seatunnel_cli/agents.py | 10 ++
seatunnel-cli/seatunnel_cli/skills/cdc_realtime.md | 39 +++++-
.../seatunnel_cli/skills/conditional_routing.md | 135 +++++++++++++++++++++
.../seatunnel_cli/skills/multi_pipeline.md | 77 ++++++++++++
4 files changed, 257 insertions(+), 4 deletions(-)
diff --git a/seatunnel-cli/seatunnel_cli/agents.py
b/seatunnel-cli/seatunnel_cli/agents.py
index 07adfc6670..1a5085c7b5 100644
--- a/seatunnel-cli/seatunnel_cli/agents.py
+++ b/seatunnel-cli/seatunnel_cli/agents.py
@@ -1108,6 +1108,16 @@ When generating configs with **multiple source/sink
blocks**:
- ALL source blocks go inside ONE `source {{ }}` section
- ALL sink blocks go inside ONE `sink {{ }}` section
+Chained/wide DAGs (transforms in the middle, 5+ blocks):
+- Wire every hop explicitly: source → transform → sink each consume the
+ upstream label and emit their own (`a_raw` → `a_filtered` → sink).
+- After inserting a transform, re-point the downstream consumer to the
+ transform's OUTPUT label — a sink still reading the source label silently
+ bypasses the transform.
+- To SPLIT one stream by condition: several transforms may consume the SAME
+ source label (each with its own predicate + distinct output label). Two
+ different sources must never emit the same label.
+
Example — Two Jdbc sources routed to Console and Assert:
```hocon
env {{
diff --git a/seatunnel-cli/seatunnel_cli/skills/cdc_realtime.md
b/seatunnel-cli/seatunnel_cli/skills/cdc_realtime.md
index 825d82e618..29901e2c5e 100644
--- a/seatunnel-cli/seatunnel_cli/skills/cdc_realtime.md
+++ b/seatunnel-cli/seatunnel_cli/skills/cdc_realtime.md
@@ -54,12 +54,42 @@ Data flows as a never-ending stream of insert/update/delete
events.
- For multi-table CDC, use `table-name` regex or create multiple CDC source
blocks.
- CDC sources need database permissions: REPLICATION SLAVE, REPLICATION CLIENT
for MySQL.
+### PostgreSQL-CDC specifics (differs from MySQL-CDC)
+
+- PostgreSQL-CDC reads the logical replication stream (WAL), which requires
+ server-side prerequisites: `wal_level = logical` on the server, and a user
+ with the REPLICATION attribute.
+- **`slot.name` should be set explicitly** — a stable replication slot name
+ (e.g. `"seatunnel_slot"`). Each concurrent CDC job needs its own slot.
+- **`decoding.plugin.name`** selects the logical decoding plugin. Use
+ `"pgoutput"` for PostgreSQL 10+ (built-in, no server install needed);
+ `"decoderbufs"` only when that plugin is installed on the server.
+- `table-names` (plural, list-typed) takes schema-qualified entries. The
+ canonical form is TWO-part `"schema.table"` (e.g. `["inventory.orders"]`)
+ — the connector's own error message calls it `schemaName.tableName`.
+ A three-part `"database.schema.table"` entry is also accepted, but the
+ leading database segment is ignored (it does NOT filter by database).
+ Database selection comes from `database-names` (plural, list-typed);
+ the connector connects to the first listed database. Always confirm
+ exact keys with `get_connector_info`.
+- Debezium engine properties (e.g. a custom publication name) are passed
+ through the nested `debezium { }` map option, such as
+ `debezium { publication.name = "my_pub" }` — never as top-level options.
+
+| Option | MySQL-CDC | PostgreSQL-CDC |
+|---|---|---|
+| stream source | binlog | logical replication (WAL) |
+| `slot.name` | n/a | recommended, one per job |
+| `decoding.plugin.name` | n/a | `pgoutput` (PG 10+) |
+| `table-names` entry format | `database.table` | `schema.table` (canonical) |
+| server prerequisite | binlog ROW mode | `wal_level = logical` |
+
## SOP
1. **Identify CDC source connector** (MySQL-CDC, PostgreSQL-CDC, etc.) from
the plan.
2. **Identify sink connector** — verify it supports upsert semantics for CDC
correctness.
3. **Set env block**: `job.mode = "STREAMING"`, `checkpoint.interval = 10000`.
-4. **Fill CDC source options**: hostname, port, username, password,
database-name, table-name. Use `get_connector_info` for exact option keys.
+4. **Fill CDC source options**: hostname, port, username, password, and the
database/table selection keys. Use `get_connector_info` for exact option keys —
they vary by connector (e.g. PostgreSQL-CDC uses plural
`database-names`/`table-names`).
5. **Fill sink options**: fetch connector metadata, ensure primary key /
upsert config is set.
6. **Set routing**: `plugin_output` on source, `plugin_input` on sink.
7. **Apply golden example** if available for this (source, sink) pair.
@@ -69,7 +99,7 @@ Data flows as a never-ending stream of insert/update/delete
events.
- MUST use `job.mode = "STREAMING"` — never BATCH for CDC.
- MUST include `checkpoint.interval` in env block.
-- CDC source option keys use hyphens: `database-name`, `table-name` — NOT
underscores.
+- CDC source option keys use hyphens (e.g. `database-name`/`database-names`) —
NOT underscores.
- Do NOT add `query` to CDC sources — they capture changelogs, not arbitrary
SQL.
- Credentials always as `${ENV_VAR}`.
@@ -88,8 +118,9 @@ source {
port = <port>
username = "${DB_USER}"
password = "${DB_PASSWORD}"
- database-name = "<database>"
- table-name = "<table>"
+ database-names = ["<database>"]
+ # MySQL-CDC entries: "<database>.<table>"; PostgreSQL-CDC:
"<schema>.<table>"
+ table-names = ["<database-or-schema-qualified-table>"]
plugin_output = "<routing_label>"
}
}
diff --git a/seatunnel-cli/seatunnel_cli/skills/conditional_routing.md
b/seatunnel-cli/seatunnel_cli/skills/conditional_routing.md
new file mode 100644
index 0000000000..eed870a82d
--- /dev/null
+++ b/seatunnel-cli/seatunnel_cli/skills/conditional_routing.md
@@ -0,0 +1,135 @@
+<!--
+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.
+-->
+
+---
+name: conditional_routing
+description: Split one source stream into multiple sinks by row-level
conditions (predicate routing)
+triggers:
+ - split
+ - route
+ - routing
+ - partition by
+ - separate
+ - filter into
+ - based on
+ - by condition
+ - by status
+ - by level
+ - greater than
+ - less than
+ - 拆分
+ - 分流
+ - 路由
+ - 大于
+ - 小于
+tools:
+ - get_connector_info
+composable: true
+---
+
+## When to Use
+
+User wants rows from ONE source delivered to DIFFERENT sinks depending on a
+row-level condition. Examples: "route amount >= 1000 to files, the rest to
+console", "ERROR logs to console, everything else to archive", "PAID orders
+here, unpaid there".
+
+This is NOT multi-pipeline (independent sources) and NOT fan-out (same rows
+to all sinks) — it is one stream split by mutually exclusive predicates.
+
+## Domain Knowledge
+
+- The split is built from **parallel SQL transforms consuming the SAME source
+ output**: each transform declares `plugin_input = "<source_label>"` and its
+ own WHERE predicate, then emits a distinct `plugin_output`.
+- SeaTunnel duplicates the source stream to every transform that references
+ its label — no extra config needed on the source.
+- Each branch then needs its own sink consuming that branch's label.
+- The Zeta SQL transform supports projection + WHERE only (no GROUP BY,
+ JOIN, or ORDER BY). Keep predicates simple comparisons/boolean logic.
+- In the SQL transform, `table_name` must reference the INPUT label (the
+ source's plugin_output), and the FROM clause uses that same name.
+
+## SOP
+
+1. **Identify the split condition** and the number of branches (usually 2).
+2. **Make predicates mutually exclusive and complete**: `amount > 100` pairs
+ with `amount <= 100` (not `amount < 100`, which silently drops equality).
+ If the user's wording leaves a gap, close it and note the choice.
+3. **One source block** with a single `plugin_output` label.
+4. **One SQL transform per branch**, all with `plugin_input` = source label,
+ each with a distinct `plugin_output` and its WHERE predicate.
+5. **One sink per branch**, each consuming its branch label.
+6. **Validate wiring**: every label consumed exists; branch labels are
+ unique; no branch left without a sink.
+
+## Constraints
+
+- Do NOT read the source twice — one source block, shared by transforms.
+- Do NOT chain the transforms (both consume the SOURCE label, not each other).
+- Predicates MUST be mutually exclusive and SHOULD cover all rows.
+- Only WHERE-level SQL (Zeta SQL transform limitation).
+
+## Pattern
+
+Split `transactions` by amount into two destinations:
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ url = "jdbc:mysql://<host>:3306/<db>"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "${MYSQL_USER}"
+ password = "${MYSQL_PASSWORD}"
+ query = "SELECT * FROM transactions"
+ plugin_output = "tx_all"
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "tx_all"
+ plugin_output = "tx_large"
+ query = "SELECT * FROM tx_all WHERE amount >= 1000"
+ }
+ Sql {
+ plugin_input = "tx_all"
+ plugin_output = "tx_small"
+ query = "SELECT * FROM tx_all WHERE amount < 1000"
+ }
+}
+
+sink {
+ LocalFile {
+ plugin_input = "tx_large"
+ path = "/data/tx_large"
+ file_format_type = "json"
+ }
+ Console {
+ plugin_input = "tx_small"
+ }
+}
+```
+
+Wiring summary: `source(tx_all) ⇒ [Sql→tx_large, Sql→tx_small] ⇒ sinks`.
+Both transforms consume `tx_all`; the two predicates are exclusive
+(`>= 1000` / `< 1000`) and together cover every row.
diff --git a/seatunnel-cli/seatunnel_cli/skills/multi_pipeline.md
b/seatunnel-cli/seatunnel_cli/skills/multi_pipeline.md
index d5d3efe77e..1bddb69875 100644
--- a/seatunnel-cli/seatunnel_cli/skills/multi_pipeline.md
+++ b/seatunnel-cli/seatunnel_cli/skills/multi_pipeline.md
@@ -64,6 +64,83 @@ Expansion is NOT needed when:
- Sink supports multi-table (e.g., Jdbc with `generate_sink_sql = true`)
- Source uses multi-table syntax natively (e.g., CDC with regex `table-name`)
+### Wide-DAG Wiring (5+ blocks, transforms in the middle)
+
+When a job mixes pipelines WITH transforms and pipelines WITHOUT them, wire
+each hop explicitly — every block consumes exactly one upstream label and
+emits its own. Always produce ONE complete config with single `source { }`,
+`transform { }`, and `sink { }` sections (never separate fragments per block).
+
+Complete example — chain `source → Sql → FieldMapper → sink` plus an
+independent second pipeline in the same job:
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ url = "jdbc:mysql://host:3306/shop"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "${MYSQL_USER}"
+ password = "${MYSQL_PASSWORD}"
+ query = "SELECT * FROM orders"
+ plugin_output = "orders_raw"
+ }
+ Jdbc {
+ url = "jdbc:mysql://host:3306/shop"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "${MYSQL_USER}"
+ password = "${MYSQL_PASSWORD}"
+ query = "SELECT * FROM audit_log"
+ plugin_output = "audit_raw"
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "orders_raw"
+ plugin_output = "orders_filtered"
+ query = "SELECT * FROM orders_raw WHERE amount > 10"
+ }
+ FieldMapper {
+ plugin_input = "orders_filtered"
+ plugin_output = "orders_final"
+ field_mapper = {
+ id = order_id
+ amount = total
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "orders_final"
+ }
+ Console {
+ plugin_input = "audit_raw"
+ }
+}
+```
+
+The chain hops are: `orders_raw` (source) → `orders_filtered` (Sql) →
+`orders_final` (FieldMapper) → first sink; the second pipeline goes
+`audit_raw` → second sink directly.
+
+Common wiring mistakes to avoid:
+- **Label reuse across chains**: one label may feed multiple parallel
+ branches (intentional split), but two different SOURCES must never emit
+ the same label.
+- **Broken chain**: a transform emitting `orders_filtered` while the sink
+ still consumes `orders_raw` silently bypasses the transform. After adding
+ a transform, re-point the downstream consumer to the transform's output
+ label.
+- **Transform applied to the wrong pipeline**: transforms only apply to the
+ chain whose label they consume — placement inside `transform { }` does not
+ scope them; only `plugin_input` does.
+
## SOP
1. **Count pipelines** from the structured plan.
