{"record":{"id":"7f77f4354d356aed","repo":"apache/flink","slug":"target-field-was-added-twice-to-input","errorCode":null,"errorMessage":"Target field {} was added twice to input {}","messagePattern":"Target field (.+?) was added twice to input (.+?)","errorType":"validation","errorClass":"InvalidSemanticAnnotationException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java","lineNumber":128,"sourceCode":"     *\n     * @param input the input of the source field\n     * @param sourceField the position in the source record\n     * @param targetField the position in the destination record\n     */\n    public void addForwardedField(int input, int sourceField, int targetField) {\n\n        Map<Integer, FieldSet> fieldMapping;\n\n        if (input != 0 && input != 1) {\n            throw new IndexOutOfBoundsException();\n        } else if (input == 0) {\n            fieldMapping = this.fieldMapping1;\n        } else {\n            fieldMapping = this.fieldMapping2;\n        }\n\n        if (isTargetFieldPresent(targetField, fieldMapping)) {\n            throw new InvalidSemanticAnnotationException(\n                    \"Target field \" + targetField + \" was added twice to input \" + input);\n        }\n\n        FieldSet targetFields = fieldMapping.get(sourceField);\n        if (targetFields != null) {\n            fieldMapping.put(sourceField, targetFields.addField(targetField));\n        } else {\n            fieldMapping.put(sourceField, new FieldSet(targetField));\n        }\n    }\n\n    private boolean isTargetFieldPresent(int targetField, Map<Integer, FieldSet> fieldMapping) {\n\n        for (FieldSet targetFields : fieldMapping.values()) {\n            if (targetFields.contains(targetField)) {\n                return true;\n            }\n        }","sourceCodeStart":110,"sourceCodeEnd":146,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java#L110-L146","documentation":"Thrown by DualInputSemanticProperties.addForwardedField() when the same target field position is already mapped from a (different) source field in the same input. Flink's two-input semantic model requires each output field to be forwarded from at most one source field per input; a duplicate target means the annotation is internally contradictory and the optimizer cannot trust it.","triggerScenarios":"Annotating a CoGroup/Join/FlatJoin function with @ForwardedFieldsSecond or @ForwardedFieldsFirst where two distinct source fields both claim to land in the same output position; manually calling addForwardedField(input, srcA, tgtX) then addForwardedField(input, srcB, tgtX).","commonSituations":"Writing @ForwardedFields annotations on two-input operators by copying from a single-input pattern without realising the second-input annotation also targets output fields; miscounting output field positions when the function emits a Tuple whose field index is reused.","solutions":["Audit the @ForwardedFieldsFirst / @ForwardedFieldsSecond annotations: each output field index must appear as a target exactly once per input.","If the function genuinely produces one field from two inputs, drop the forwarded-field annotation for that target and use @ReadFields or no annotation instead.","Check the exception's input number (0 or 1) and targetField value to localise which annotation is duplicated."],"exampleFix":"// before\n@ForwardedFieldsFirst(\"0->0\")\n@ForwardedFieldsSecond(\"0->0\") // both claim output field 0\npublic class MyJoin extends RichJoinFunction<Tuple2<String,Integer>, Tuple2<String,Integer>, Tuple2<String,Integer>> { ... }\n// after — only one source legitimately forwards to output field 0\n@ForwardedFieldsFirst(\"0->0;1->1\")\npublic class MyJoin extends RichJoinFunction<...> { ... }","handlingStrategy":"validation","validationCode":"// before annotating, ensure each output field appears as a target at most once per input\nSet<Integer> targetsForInput0 = new HashSet<>();\nfor (ForwardedFieldSpec s : specsFirstInput) {\n    if (!targetsForInput0.add(s.target)) throw new IllegalStateException(\"duplicate target \" + s.target);\n}","typeGuard":null,"tryCatchPattern":"try {\n    props.addForwardedField(input, src, tgt);\n} catch (InvalidSemanticAnnotationException e) {\n    // annotations are static metadata; fix the annotation rather than catch at runtime\n    throw new IllegalStateException(\"Bad @ForwardedFields annotation: \" + e.getMessage(), e);\n}","preventionTips":["Each output field index must be a forwarded target of at most one source field per input.","Validate forwarded-field annotations in a unit test that constructs the SemanticProperties.","When copying single-input annotations to a two-input operator, re-check the second-input targets."],"tags":["flink-core","semantic-annotations","forwarded-fields","two-input","dataset-api"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}