{"record":{"id":"3d00d11a8e1b234c","repo":"apache/iceberg","slug":"field-fieldindex-in-target-schema-targettype","errorCode":null,"errorMessage":"Field ${fieldIndex} in target schema ${targetType} is non-nullable but does not exist in source schema.","messagePattern":"Field (.+?) in target schema (.+?) is non-nullable but does not exist in source schema\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java","lineNumber":153,"sourceCode":"  class RowDataConverter implements DataConverter {\n    private final RowData.FieldGetter[] fieldGetters;\n    private final DataConverter[] dataConverters;\n\n    RowDataConverter(RowType sourceType, RowType targetType) {\n      this.fieldGetters = new RowData.FieldGetter[targetType.getFields().size()];\n      this.dataConverters = new DataConverter[targetType.getFields().size()];\n\n      for (int i = 0; i < targetType.getFields().size(); i++) {\n        RowData.FieldGetter fieldGetter;\n        DataConverter dataConverter;\n        RowType.RowField targetField = targetType.getFields().get(i);\n        int sourceFieldIndex = sourceType.getFieldIndex(targetField.getName());\n        if (sourceFieldIndex == -1) {\n          if (targetField.getType().isNullable()) {\n            fieldGetter = row -> null;\n            dataConverter = value -> null;\n          } else {\n            throw new IllegalArgumentException(\n                String.format(\n                    \"Field %s in target schema %s is non-nullable but does not exist in source schema.\",\n                    i + 1, targetType));\n          }\n        } else {\n          RowType.RowField sourceField = sourceType.getFields().get(sourceFieldIndex);\n          fieldGetter = RowData.createFieldGetter(sourceField.getType(), sourceFieldIndex);\n          dataConverter = DataConverter.getNullable(sourceField.getType(), targetField.getType());\n        }\n\n        this.fieldGetters[i] = fieldGetter;\n        this.dataConverters[i] = dataConverter;\n      }\n    }\n\n    @Override\n    public RowData convert(Object object) {\n      RowData sourceData = (RowData) object;","sourceCodeStart":135,"sourceCodeEnd":171,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java#L135-L171","documentation":"RowDataConverter matches target schema fields to source fields by name. If a target field is absent from the source schema AND the target field is non-nullable, there is no valid value to produce, so it throws IllegalArgumentException. Nullable missing fields are tolerated (filled with null).","triggerScenarios":"Schema evolution where the target schema adds a required (non-nullable, no default) field that doesn't exist in the source RowData schema being converted.","commonSituations":"Evolving an Iceberg table by adding a required column with required=true and no write default, then writing old records through the Flink dynamic sink; mismatched schemas passed to the dynamic sink.","solutions":["Add the new column as nullable, or with a write default, instead of required (ALTER TABLE ... ADD COLUMN c X  -- nullable).","Provide a default value for the new field so old data can be converted (ADD COLUMN ... AFTER ... with a write-default).","Backfill the source data so every record contains the new field before switching the sink to the evolved schema."],"exampleFix":"// before\nALTER TABLE db.t ADD COLUMN region STRING NOT NULL;\n// after\nALTER TABLE db.t ADD COLUMN region STRING; -- nullable, or provide write-default","handlingStrategy":"validation","validationCode":"for (Types.NestedField f : targetSchema.columns()) {\n  if (f.isRequired() && sourceSchema.findField(f.name()) == null) {\n    throw new IllegalArgumentException(\"New required field without default: \" + f.name());\n  }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Add columns as nullable or with write defaults when evolving schemas.","Backfill source data before making fields required.","Run schema-diff checks between source and target schemas before writing."],"tags":["flink","dynamic-sink","schema-evolution","nullability"],"backgroundTag":"schema-validation-failed","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}