{"record":{"id":"74d3aadd7941a8ae","repo":"apache/seatunnel","slug":"incremental-snapshot-for-tables-requires-primary-k-74d3aa","errorCode":null,"errorMessage":"Incremental snapshot for tables requires primary key, but table %s doesn't have primary key.","messagePattern":"Incremental snapshot for tables requires primary key, but table (.+?) doesn't have primary key\\.","errorType":"validation","errorClass":"SeaTunnelException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java","lineNumber":326,"sourceCode":"                }\n            } else {\n                for (int i = 0; i < primaryKeyNum; i++) {\n                    statement.setObject(i + 1, splitStart[i]);\n                    statement.setObject(i + 1 + primaryKeyNum, splitEnd[i]);\n                    statement.setObject(i + 1 + 2 * primaryKeyNum, splitEnd[i]);\n                }\n            }\n            return statement;\n        } catch (Exception e) {\n            throw new RuntimeException(\"Failed to build the split data read statement.\", e);\n        }\n    }\n\n    public static SeaTunnelRowType getSplitType(\n            Table table, RelationalDatabaseConnectorConfig dbzConnectorConfig) {\n        List<Column> primaryKeys = table.primaryKeyColumns();\n        if (primaryKeys.isEmpty()) {\n            throw new SeaTunnelException(\n                    String.format(\n                            \"Incremental snapshot for tables requires primary key,\"\n                                    + \" but table %s doesn't have primary key.\",\n                            table.id()));\n        }\n\n        // use first field in primary key as the split key\n        return getSplitType(primaryKeys.get(0), dbzConnectorConfig);\n    }\n\n    public static BinlogOffset getBinlogPosition(SourceRecord dataRecord) {\n        return getBinlogPosition(dataRecord.sourceOffset());\n    }\n\n    public static BinlogOffset getBinlogPosition(Map<String, ?> offset) {\n        Map<String, String> offsetStrMap = new HashMap<>();\n        for (Map.Entry<String, ?> entry : offset.entrySet()) {\n            offsetStrMap.put(","sourceCodeStart":308,"sourceCodeEnd":344,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java#L308-L344","documentation":"MySqlUtils.getSplitType derives the SeaTunnelRowType used for chunk splitting from the table's primary key, and throws a SeaTunnelException when the table has no primary key, because incremental snapshot chunking requires a unique, ordered key. The library fails fast instead of producing incorrect or duplicated snapshot chunks.","triggerScenarios":"Calling getSplitType for a table whose primaryKeyColumns() is empty while incremental snapshot is enabled; typically a table created without PRIMARY KEY or with only a non-clustered unique key the connector doesn't treat as PK.","commonSituations":"Legacy tables without PKs added to a CDC table list; replica tables stripped of PKs; user assumes a UNIQUE index suffices but connector requires an actual primary key.","solutions":["Add a primary key to the table (ALTER TABLE ... ADD PRIMARY KEY) and restart the job","If adding a PK is impossible, exclude the table from the incremental-snapshot table list or use a non-incremental snapshot mode","If a suitable unique NOT NULL column exists, add a PK on it","Check table.id() in the message to identify exactly which table lacks the key"],"exampleFix":"// before\nCREATE TABLE orders (id BIGINT, name VARCHAR(64));\n// after\nALTER TABLE orders ADD PRIMARY KEY (id);","handlingStrategy":"validation","validationCode":"ResultSet rs = st.executeQuery(\n    \"SELECT COUNT(*) FROM information_schema.table_constraints \" +\n    \"WHERE table_schema=DATABASE() AND table_name='t' AND constraint_type='PRIMARY KEY'\");\nif (!rs.next() || rs.getLong(1) == 0) throw new IllegalStateException(\"PK required for incremental snapshot\");","typeGuard":null,"tryCatchPattern":"try {\n    SeaTunnelRowType splitType = MySqlUtils.getSplitType(table, dbzConfig);\n} catch (SeaTunnelException e) {\n    // fall back to non-incremental snapshot or exclude table\n}","preventionTips":["Enforce PKs on all tables registered for CDC","Audit table lists with SHOW CREATE TABLE before enabling incremental snapshot","Fall back to non-incremental mode for PK-less legacy tables","Remember UNIQUE indexes don't substitute for a PK here"],"tags":["mysql","cdc","primary-key","incremental-snapshot","config"],"backgroundTag":"missing-required-config-field","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}