{"record":{"id":"29ddace1cda95df9","repo":"apache/seatunnel","slug":"incremental-snapshot-for-tables-requires-primary-k-29ddac","errorCode":null,"errorMessage":"Incremental snapshot for tables requires primary key, but table ${table.id()} doesn't have primary key.","messagePattern":"Incremental snapshot for tables requires primary key, but table (.+?) doesn't have primary key\\.","errorType":"exception","errorClass":"SeaTunnelException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java","lineNumber":265,"sourceCode":"                        chunkSize);\n        return jdbc.prepareQueryAndMap(\n                query,\n                ps -> ps.setObject(1, includedLowerBound),\n                rs -> {\n                    if (!rs.next()) {\n                        // this should never happen\n                        throw new SQLException(\n                                String.format(\n                                        \"No result returned after running query [%s]\", query));\n                    }\n                    return rs.getObject(1);\n                });\n    }\n\n    public static SeaTunnelRowType getSplitType(Table table) {\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));\n    }\n\n    public static SeaTunnelRowType getSplitType(Column splitColumn) {\n        return new SeaTunnelRowType(\n                new String[] {splitColumn.name()},\n                new SeaTunnelDataType<?>[] {PostgresTypeUtils.convertFromColumn(splitColumn)});\n    }\n\n    public static Offset getLsnPosition(SourceRecord record) {\n        return getLsnPosition(record.sourceOffset());","sourceCodeStart":247,"sourceCodeEnd":283,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java#L247-L283","documentation":"PostgresUtils.getSplitType converts the table's primary key columns into the split key type used by the incremental snapshot framework. Incremental snapshots chop tables into chunks by a key column, so if table.primaryKeyColumns() is empty it throws SeaTunnelException stating the table must have a primary key.","triggerScenarios":"Enabling incremental snapshot (scan.incremental.snapshot.enabled=true, the default) on a Postgres table defined without a PRIMARY KEY constraint, or where the PK is not visible to the schema reader (e.g. PK excluded from publication/schema read).","commonSituations":"Legacy Postgres tables created without PKs; tables whose PK was dropped for bulk-load performance; partitioned tables where the PK lives on partitions; readers with insufficient privilege to see constraints.","solutions":["Add a primary key to the table: ALTER TABLE <t> ADD PRIMARY KEY (<cols>)","Choose an existing NOT NULL UNIQUE column as the effective key if a PK cannot be added","Disable incremental snapshot (scan.incremental.snapshot.enabled=false) to fall back to the lock-based snapshot mode, if supported for your version","Verify the connector's schema read can see the constraint (privileges/catalog visibility)"],"exampleFix":"// before\nCREATE TABLE users (id bigint, name text);\n// after\nALTER TABLE users ADD PRIMARY KEY (id);","handlingStrategy":"validation","validationCode":"SELECT a.attname FROM pg_index i\nJOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)\nWHERE i.indrelid = 'public.orders'::regclass AND i.indisprimary;","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always define a PRIMARY KEY on CDC tables","Do not drop PKs for bulk loads on replicated tables","Fall back to non-incremental snapshot mode only when the table truly has no key"],"tags":["postgres","cdc","primary-key","incremental-snapshot"],"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"}