{"record":{"id":"f41800bf5f6bf3b9","repo":"apache/iceberg","slug":"field-s-in-target-schema-s-is-non-nullable-but-d","errorCode":null,"errorMessage":"Field %s in target schema %s 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/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java#L135-L171","documentation":"RowDataConverter maps target schema fields to source schema fields by name. If a target field is missing from the source and the target field is nullable, the converter substitutes null; but if the target field is non-nullable, it throws this IllegalArgumentException because a required value cannot be produced. This enforces schema evolution safety: non-nullable target columns must exist in the incoming data.","triggerScenarios":"Evolving the target table by adding a REQUIRED (non-nullable) column that does not exist in the incoming Flink RowData schema, then writing through the dynamic sink.","commonSituations":"ALTER TABLE ADD COLUMN without making the new column nullable; source stream schema not updated after a target schema migration; backfilling old records against an evolved target schema.","solutions":["Make the new target column nullable so missing source values map to null","Update the source RowData schema/data to include the new field with a value","Use a schema evolution with a default value (Iceberg add-column with write-default/initial-default) if the table format supports it"],"exampleFix":"// before\n// target schema: RequiredField LONG (required), absent in source\n// after\n// evolve target so the new field is nullable, or supply the field in the source RowType","handlingStrategy":"validation","validationCode":"for (RowType.RowField targetField : targetRowType.getFields()) {\n  if (!targetField.getType().isNullable()\n      && sourceRowType.getFieldIndex(targetField.getName()) == -1) {\n    throw new IllegalArgumentException(\n        \"Non-nullable target field missing in source: \" + targetField.getName());\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  converter = new RowDataConverter(sourceRowType, targetRowType);\n} catch (IllegalArgumentException e) {\n  // evolve schema or update source before writing\n}","preventionTips":["Make all ALTER TABLE ADD COLUMN targets nullable unless the pipeline guarantees values","Keep source stream schema and target table schema in sync through schema evolution","Validate schema compatibility at job startup before writing"],"tags":["flink","schema-evolution","nullable"],"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"}