{"record":{"id":"92c67905af2fa4ac","repo":"apache/seatunnel","slug":"primary-key-s-is-not-in-table-s-columns-s","errorCode":null,"errorMessage":"Primary key(%s) is not in table(%s) columns(%s)","messagePattern":"Primary key\\((.+?)\\) is not in table\\((.+?)\\) columns\\((.+?)\\)","errorType":"exception","errorClass":"java.lang.IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/CatalogTableUtils.java","lineNumber":76,"sourceCode":"                        catalogTableConfig.getTable());\n            } else {\n                log.warn(\n                        \"Table {} is not found in catalog tables, skip to merge config\",\n                        catalogTableConfig.getTable());\n            }\n        }\n        return new ArrayList<>(catalogTableMap.values());\n    }\n\n    public static CatalogTable mergeCatalogTableConfig(\n            final CatalogTable table, JdbcSourceTableConfig config) {\n        List<String> columnNames =\n                table.getTableSchema().getColumns().stream()\n                        .map(c -> c.getName())\n                        .collect(Collectors.toList());\n        for (String pk : config.getPrimaryKeys()) {\n            if (!columnNames.contains(pk)) {\n                throw new IllegalArgumentException(\n                        String.format(\n                                \"Primary key(%s) is not in table(%s) columns(%s)\",\n                                pk, table.getTablePath(), columnNames));\n            }\n        }\n        PrimaryKey primaryKeys =\n                PrimaryKey.of(\n                        \"pk\" + (config.getPrimaryKeys().hashCode() & Integer.MAX_VALUE),\n                        config.getPrimaryKeys());\n        List<Column> columns =\n                table.getTableSchema().getColumns().stream()\n                        .map(\n                                column -> {\n                                    if (config.getPrimaryKeys().contains(column.getName())\n                                            && column.isNullable()) {\n                                        log.warn(\n                                                \"Primary key({}) is nullable for catalog table {}\",\n                                                column.getName(),","sourceCodeStart":58,"sourceCodeEnd":94,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/CatalogTableUtils.java#L58-L94","documentation":"CatalogTableUtils.mergeCatalogTableConfig validates that every primary key listed in the user-supplied CDC table config exists in the target table's schema columns. If a configured PK is not among the table's actual column names, it throws IllegalArgumentException naming the key, the table path, and the available columns. This prevents building a PrimaryKey that cannot be mapped onto the source table.","triggerScenarios":"table-names/primary-keys config in a CDC source declares a primaryKeys entry that does not match any column name of the captured table (case mismatch, renamed/dropped column, wrong table).","commonSituations":"Case-sensitivity: table columns are lowercase but primaryKeys config is uppercase; column renamed upstream after the config was written; copying a config between environments where schemas differ; specifying PKs for a table without checking its DDL.","solutions":["Compare the configured primaryKeys against the column list printed in the error and fix the names to match exactly (including case).","Remove the primaryKeys override if the table already has a real primary key; the connector will pick it up from the catalog.","Run a schema query (SHOW CREATE TABLE / information_schema) to confirm current column names before configuring.","If the column was renamed, update both the table config and any downstream key-based processing."],"exampleFix":"// before\nprimaryKeys = [\"Id\"]   // table column is \"id\"\n// after\nprimaryKeys = [\"id\"]","handlingStrategy":"validation","validationCode":"// validate PK config against actual schema before job start\nList<String> columns = table.getTableSchema().getColumns()\n        .stream().map(c -> c.getName()).collect(Collectors.toList());\nfor (String pk : config.getPrimaryKeys()) {\n    if (!columns.contains(pk)) {\n        throw new IllegalArgumentException(\"Unknown PK: \" + pk);\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    CatalogTableUtils.mergeCatalogTableConfig(table, config);\n} catch (IllegalArgumentException e) {\n    log.error(\"Fix primaryKeys config: {}\", e.getMessage());\n}","preventionTips":["Copy column names exactly (respecting case) from SHOW CREATE TABLE.","Re-verify configs after any upstream schema migration.","Prefer letting the connector discover PKs from the catalog when one exists."],"tags":["cdc","configuration","primary-key","schema-mismatch"],"backgroundTag":"schema-validation-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}