li3zhi4 opened a new issue, #11742: URL: https://github.com/apache/seatunnel/issues/11742
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue) and found no issue/PR covering composite-key **multi-column chunk splitting** in the MySQL CDC snapshot phase: - #9949 "[Bug] [mysql-cdc] When mysql-cdc retrieves the partitioning column, if there is a composite index, it will retrieve fields other than the primary key and index columns as the partitioning column, leading to slow SQL" (closed) — about **split-column selection** (which single column to pick), fixed by #10236/#10319; this issue is about splitting on **multiple** columns. - #7794 "[Bug] [MYSQL-CDC] ... 不使用主键(id)进行分割，而是使用唯一键" (closed) — split-column selection again, single-column splitting. - #10596 / #10604 "[Feature][CDC/JDBC] Add option to disable sampling-based sharding" (closed/merged) — adds the `split.allow-sampling=false` toggle that this issue's scenario relies on (single-column split degrades when sampling is off and the first key column repeats heavily); related prerequisite, but that work is about the sampling switch itself, not multi-column composite-key splitting. - #11396 "[Fix][Connector-V2][Jdbc] Fix DATE partition column splitting" (open) — single-column DATE type handling. - #9221 "[Question] [Oracle JDBC] 同步oracle 11g失败" (closed) — usage question, mentions queryMinMax only as stack-trace noise. - #11741 "[Feature][Connector-JDBC] Support composite primary key (multi-column) in JDBC source chunk splitting" (open, 2026-08-10) — the JDBC-source counterpart of this same capability gap; submitted separately because `connector-jdbc` and the CDC connectors have independent splitters. This issue is the CDC side. ### What happened The MySQL CDC snapshot phase splits a table into chunks using **only the first column of the primary key**: ```java // MySqlUtils.getSplitColumn(Table) // use first field in primary key as the split key return primaryKeys.get(0); ``` For tables with a **composite primary key** (`PRIMARY KEY (col1, col2, ...)`), this means the whole snapshot split is driven by the first key column alone. The failure modes depend on how many distinct values the first column actually has: 1. **First column has (almost) one distinct value** — the worst case. `splitTableIntoChunks` hits `min.equals(max)` and returns a **single full-table chunk** (`ChunkRange.all()`), so the whole table is read by one reader: parallelism is lost entirely and a large table's snapshot takes as long as a single-threaded scan. Real-world example: big tables whose first PK column is a constant/very-low-cardinality code (the first key column repeats heavily). 2. **First column has low cardinality / many duplicates** — `splitUnevenlySizedChunks` walks the boundary with `nextChunkEnd()`, but whenever `queryNextChunkMax` returns a boundary equal to the previous one it must call `queryMin` to skip past the duplicated values. With heavy duplicates the chunk count collapses and remaining chunks are severely skewed: a single chunk can dominate the snapshot duration while others cover almost nothing. 3. The chunk scan SQL (`buildSplitQuery` / `addPrimaryKeyColumnsToCondition`) ANDs per-column predicates (`col1 <= ? AND col2 <= ?`) — a bounding-box comparison, not a lexicographic tuple comparison — which can wrongly include/exclude rows exactly at a chunk boundary, causing boundary rows to be duplicated or missed once multi-column split keys are introduced. 4. `MySqlUtils.queryMinMax` uses `MIN()` / `MAX()` aggregates over a single column; for a composite key this either scans without index benefit or (in the multi-column path) is not implemented at all. SeaTunnel's base layer (`AbstractJdbcSourceChunkSplitter` / `JdbcSourceChunkSplitter`) has no multi-column path on `dev` — the entire splitter is single-column (`getSplitColumn` returns one `Column`, `queryMinMax(jdbc, tableId, Column)`), so composite-key tables cannot benefit from tuple-ordered splitting. ### SeaTunnel Version Current upstream `dev` (verified 2026-08-10 by code inspection); the same single-column split behavior exists in 2.3.13. ### SeaTunnel Config ```hocon source { MySQL-CDC { parallelism = 4 url = "jdbc:mysql://<host>:3306/<db>" username = "user" password = "pass" database-names = ["<db>"] table-names = ["<db>.<composite_key_table>"] } } ``` A table like: ```sql CREATE TABLE t ( order_id BIGINT, line_no INT, payload TEXT, PRIMARY KEY (order_id, line_no) ); ``` ### Root cause (confirmed by code inspection) - `MySqlUtils.getSplitColumn()` hard-codes `primaryKeys.get(0)` as the split key — multi-column keys are never considered. - `AbstractJdbcSourceChunkSplitter` / `JdbcSourceChunkSplitter` on `dev` expose only single-column `queryMinMax` / `queryMin` / `queryNextChunkMax`; there is no `List<Column>` API to drive tuple-ordered splitting. Verified 2026-08-10: **all** ChunkSplitter implementations on `dev` (MySQL, Oracle, PostgreSQL, SqlServer, Mongo, JDBC and the base classes) contain **zero** `List<Column>` multi-column APIs — every database is single-column split today, so this is a **common capability gap of the whole CDC/JDBC splitter layer**, not a MySQL-specific defect. - `MySqlUtils.buildSplitQuery()`'s `addPrimaryKeyColumnsToCondition` uses per-column AND predicates (bounding box), not lexicographic tuple conditions. ### Scope of application (which tables are affected) The multi-column split path is triggered by the split-key selection logic in `AbstractJdbcSourceChunkSplitter.getSplitColumns()`: | Table shape | Split key selection | Path taken | |---|---|---| | **Composite PK** (`PRIMARY KEY (a, b, ...)`, ≥2 columns), no user-configured split column | **All** PK columns (`pkColumnNames.size() > 1 || isEvenlySplitColumn(column)` — for composite keys every column is included, **no type filter**) | **Multi-column tuple splitting** (this feature) — `queryMinMax(List<Column>)` + expanded OR/AND `queryNextChunkMax` | | Single PK (1 column) | Only if the column is an evenly-split type (`TINYINT`/`SMALLINT`/`INT`/`BIGINT`/`DECIMAL`/`STRING`) | Unchanged single-column path | | User-configured split column (`split.column` / `snapshotSplitColumn`) | Always wins — single column, must be a unique key with an evenly-split type | Unchanged single-column path (composite splitting **not** applied) | | No PK | First unique-key column (evenly-split type), else single full-table split | Unchanged | Consequences worth stating explicitly: 1. **Composite PK tables are the only ones that switch to multi-column splitting** — and for them the current single-column behavior (`primaryKeys.get(0)` only) is what produces the single-chunk / skewed-chunk degradation described above. 2. **No type filter for composite keys**: a composite key whose columns include non-numeric types (e.g. `VARCHAR`) still enters the multi-column path — the expanded OR/AND form and `ORDER BY ... LIMIT 1` work for any comparable type, unlike the single-column `isEvenlySplitColumn` gate. 3. **Everything else keeps its exact current behavior** on `dev` (single PK, no PK, user-configured column) — the base-API additions are defaulted/delegating for `size() == 1`, so the change is opt-in per table shape and backward compatible. 4. The JDBC source (`connector-jdbc`) has the same shape of gap (its `findSplitKey` also picks a single PK column), handled in the separate JDBC issue #11741. ### Design rationale (why this is the right approach) 1. **Tuple-ordered splitting is the correct semantics for a composite key** — chunk boundaries must follow the key's lexicographic order, otherwise ranges overlap or leave gaps. With a single-column split key the lexicographic order is trivial, but for a composite key `(col1, col2, ...)` the correct boundary condition is a tuple comparison (`(col1, col2) > (v1, v2)`), not per-column AND predicates. Splitting on the full key tuple is therefore the direct, correctness-first consequence of having a composite PK. 2. **Expanded OR/AND form is the index-friendly way to express tuple comparisons** — a row-constructor comparison `(a, b) >= (?, ?)` often makes MySQL's optimizer fall back to a full scan, while the equivalent expanded form `a > ? OR (a = ? AND b >= ?)` lets the composite index be used. This is a standard SQL technique for multi-column range scans, so the fix is not a workaround but the canonical SQL shape. 3. **Directly complements #10604** — with the sampling switch (`split.allow-sampling=false`) now available, tables that are unevenly distributed go straight to `splitUnevenlySizedChunks`; for composite-PK tables whose first column repeats heavily, only tuple-ordered multi-column splitting keeps chunks balanced. This issue is the natural next step after that switch. ### Proposed fix **Scope decision**: the fix is layered — a generic multi-column API in the base layer (benefits every future dialect) plus the **MySQL implementation as the first dialect**. JDBC (`connector-jdbc`) is a separate module with its own splitter (`DynamicChunkSplitter`) and is **not** part of this change — on `dev` it is equally single-column (`findSplitKey` returns the first supported PK column, `queryNextChunkMax` takes a single `String splitColumnName`; verified 2026-08-10), handled in the separate JDBC issue #11741. Oracle/PostgreSQL/SqlServer splitters stay on the single-column path (their behavior on `dev` is unchanged); adding multi-column support to them is a follow-up per dialect (each needs its own tuple-comparison SQL syntax). 1. **Base layer** (`connector-cdc-base`): add multi-column split APIs (`queryMinMax / queryMin / queryNextChunkMax` accepting `List<Column>`) to `JdbcSourceChunkSplitter` / `AbstractJdbcSourceChunkSplitter`, and drive chunk iteration over the full composite key. Default implementations delegate to the single-column versions for `size() == 1`, so existing single-column dialects are unaffected. 2. **MySQL implementation** (`connector-cdc-mysql` — first dialect, the one this issue targets): - `queryMinMax(List<Column>)`: `SELECT col1, col2 FROM t ORDER BY col1 ASC, col2 ASC LIMIT 1` / `... DESC LIMIT 1` (no `MIN()/MAX()` aggregate, empty table returns null). - `queryMin(List<Column>, excludedLowerBound)` and `queryNextChunkMax(...)`: tuple comparison **expanded into OR/AND form** (`col1 > ? OR (col1 = ? AND col2 >= ?)`) so MySQL can use the composite index (row-constructor `(a,b) >= (?,?)` often cannot). - `buildSplitQuery`: replace the bounding-box `addPrimaryKeyColumnsToCondition` with a lexicographic tuple condition (`addPrimaryKeyTupleCondition`) for the data-scan SQL. 3. **Chunk boundary semantics**: half-open `[splitStart, splitEnd)` per chunk — the `NOT (= splitEnd)` exclusion bound to `splitEnd` (the `dev` binding is already correct here), so each chunk covers `[splitStart, splitEnd)` and boundary values are never lost or duplicated. ### Verification Unit tests (composite `findSplitKey` selection, multi-column `queryMinMax`/`queryMin`/`queryNextChunkMax` incl. empty table, expanded OR/AND SQL shape, half-open `[splitStart, splitEnd)` boundary semantics) plus an official MySQL CDC E2E on `dev` with a composite-PK table asserting PK-level source/target consistency — to be included in the PR. ### Zeta or Flink or Spark Version zeta ### Java or Scala Version Java 8 / 11 (both OK) ### Screenshots _No response_ ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! (port to `dev` in progress) ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
