{"record":{"id":"bec6a863b2cd6f1f","repo":"apache/seatunnel","slug":"transform-sink-don-t-support-processing-data-with","errorCode":null,"errorMessage":"Transform/Sink don't support processing data with two different structures.","messagePattern":"Transform/Sink don't support processing data with two different structures\\.","errorType":"validation","errorClass":"JobDefineCheckException","httpStatus":null,"severity":"error","filePath":"seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java","lineNumber":605,"sourceCode":"            } catch (UnsupportedOperationException e) {\n                // TODO remove it when all connector use `getProducedCatalogTables`\n                return ((SourceAction<?, ?, ?>) action).getSource().getProducedType();\n            }\n        } else if (action instanceof TransformAction) {\n            return ((TransformAction) action)\n                    .getTransform()\n                    .getProducedCatalogTable()\n                    .getSeaTunnelRowType();\n        }\n        throw new UnsupportedOperationException();\n    }\n\n    public static void checkProducedTypeEquals(Set<Action> inputActions) {\n        SeaTunnelDataType<?> expectedType = getProducedType(new ArrayList<>(inputActions).get(0));\n        for (Action action : inputActions) {\n            SeaTunnelDataType<?> producedType = getProducedType(action);\n            if (!expectedType.equals(producedType)) {\n                throw new JobDefineCheckException(\n                        \"Transform/Sink don't support processing data with two different structures.\");\n            }\n        }\n    }\n\n    @Deprecated\n    private static <T> T findLast(LinkedHashMap<?, T> map) {\n        int size = map.size();\n        int i = 1;\n        for (T value : map.values()) {\n            if (i == size) {\n                return value;\n            }\n            i++;\n        }\n        // never execution\n        return null;\n    }","sourceCodeStart":587,"sourceCodeEnd":623,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java#L587-L623","documentation":"checkProducedTypeEquals validates that all input actions feeding one transform or sink produce identical SeaTunnelDataTypes. A transform/sink merges multiple upstream inputs only when their row schemas match; two different structures cannot be combined, so parsing fails with this JobDefineCheckException.","triggerScenarios":"parseTransform or parseSink receives a Set of input actions (multi-table source or multiple upstream branches) whose produced CatalogTable types differ (different columns or types), detected during DAG construction.","commonSituations":"Connecting a multi-table source (table_path matching several tables with different schemas) directly to one sink; merging two branches of different schemas into one transform; upstream schema drift after a table was altered.","solutions":["Ensure all inputs share the same schema, or split the sink into one sink per structure.","Insert a transform (e.g. Sql/FieldMapper/Replace) to normalize each branch to a common schema before merging.","Restrict a multi-table source with table_names/table_path regex so only same-structure tables feed the sink.","Verify upstream table schemas haven't drifted; refresh cached catalogs if needed."],"exampleFix":"// before: one sink fed by multi-table source with differing schemas\nsource { Jdbc { table_path = \"db.*\" } }\nsink { Console {} }\n// after: group by compatible table sets\nsource { Jdbc { table_path = \"db.orders_.*\" } }\nsink { Console {} }","handlingStrategy":"validation","validationCode":"// ensure all inputs to a sink share one schema\nCatalogTable first = inputs.get(0).getCatalogTable();\nfor (Action a : inputs) {\n  if (!first.getTableSchema().equals(getProducedType(a).getTableSchema())) {\n    throw new IllegalArgumentException(\"Divergent input schemas feeding one sink\");\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  engine.submitJob(config);\n} catch (JobDefineCheckException e) {\n  if (e.getMessage().contains(\"two different structures\")) {\n    // split sinks or add normalizing transforms\n  }\n}","preventionTips":["Match multi-table source patterns to same-schema tables only","Insert FieldMapper/Sql transforms to normalize branches before merging","Watch upstream DDL changes that alter produced schemas"],"tags":["job-parsing","schema-mismatch","data-types","dag-construction"],"backgroundTag":"schema-mismatch","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}