{"record":{"id":"446469da9e155636","repo":"apache/seatunnel","slug":"location-plugintype-configindex-factoryid-do","errorCode":null,"errorMessage":"${location(pluginType, configIndex, factoryId)} does not support processing inputs with different schemas. Expected table ${expected.tableId} but found table ${catalogTable.tableId}.","messagePattern":"(.+?) does not support processing inputs with different schemas\\. Expected table (.+?) but found table (.+?)\\.","errorType":"validation","errorClass":"ConfigCheckException","httpStatus":null,"severity":"error","filePath":"seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunConnectValidator.java","lineNumber":429,"sourceCode":"        }\n        ReadonlyConfig schemaConfig = ReadonlyConfig.fromMap(schemaMap);\n        return schemaConfig.getOptional(ColumnOptions.COLUMNS).isPresent()\n                || entryConfig.getOptional(FieldOptions.FIELDS).isPresent()\n                || schemaConfig.getOptional(ColumnOptions.METADATA_TABLE_ID).isPresent();\n    }\n\n    private void checkCatalogTableTypesEqual(\n            List<CatalogTable> catalogTables,\n            PluginType pluginType,\n            int configIndex,\n            String factoryId) {\n        if (catalogTables.isEmpty()) {\n            return;\n        }\n        CatalogTable expected = catalogTables.get(0);\n        for (CatalogTable catalogTable : catalogTables) {\n            if (!expected.getSeaTunnelRowType().equals(catalogTable.getSeaTunnelRowType())) {\n                throw new ConfigCheckException(\n                        location(pluginType, configIndex, factoryId)\n                                + \" does not support processing inputs with different schemas. \"\n                                + \"Expected table \"\n                                + expected.getTableId()\n                                + \" but found table \"\n                                + catalogTable.getTableId()\n                                + \".\");\n            }\n        }\n    }\n\n    private List<String> getInputIds(ReadonlyConfig config) {\n        return config.getOptional(PLUGIN_INPUT).orElse(Collections.singletonList(DEFAULT_ID));\n    }\n\n    private <T> T findLast(LinkedHashMap<?, T> map) {\n        if (map.isEmpty()) {\n            throw new ConfigCheckException(","sourceCodeStart":411,"sourceCodeEnd":447,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/DryRunConnectValidator.java#L411-L447","documentation":"checkCatalogTableTypesEqual verifies that all CatalogTables feeding a plugin (a transform's inputs or a sink's merged inputs) have identical SeaTunnelRowType schemas. If any table's row type differs from the first table's, a ConfigCheckException is thrown reporting the expected and mismatching TableIds. This prevents plugins from receiving inputs with incompatible column layouts.","triggerScenarios":"During validateTransform or resolveSinkInputTables when a plugin declares multiple plugin_input upstreams whose inferred schemas differ in column names, types, or order — e.g. two JDBC sources reading tables with different columns merged into one sink.","commonSituations":"Union-style configs joining heterogeneous tables into one sink; upstream schema drift after a source table was altered; copy-pasted source blocks with different column lists.","solutions":["Align the schemas: make each upstream select/project the same columns with the same types and order.","Insert a transform (e.g. SQL/FieldMapper) per branch to normalize columns before they converge.","If a source table changed, update the config or pin the query to an explicit column list."],"exampleFix":"// before\nsource {\n  Jdbc { query = \"select id, name from users\"; plugin_output = \"a\" }\n  Jdbc { query = \"select id, email from users2\"; plugin_output = \"b\" }\n}\n// after\nsource {\n  Jdbc { query = \"select id, name from users\"; plugin_output = \"a\" }\n  Jdbc { query = \"select id, name from users2\"; plugin_output = \"b\" }\n}","handlingStrategy":"validation","validationCode":"// compare inferred schemas of all inputs to a plugin before validating\nconst schemas = inputs.map(i => i.seaTunnelRowType);\nconst first = JSON.stringify(schemas[0]);\nif (schemas.some(s => JSON.stringify(s) !== first)) {\n  throw new Error(\"Inputs have different schemas; normalize with a transform before merging\");\n}","typeGuard":null,"tryCatchPattern":"try { validateConf(conf, DryRun.CONNECT); } catch (ConfigCheckException e) { if (e.getMessage().contains(\"does not support processing inputs with different schemas\")) { addSchemaNormalizingTransforms(conf); } else { throw e; } }","preventionTips":["Use explicit column lists in source queries so schemas stay stable","Insert FieldMapper/SQL transforms to align columns before they converge","Re-run dry-run validation after any upstream table DDL change"],"tags":["dry-run","schema-mismatch","transform","sink"],"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-14T11:17:12.474Z"}