diff --git a/core/src/main/java/io/onetable/iceberg/IcebergSchemaExtractor.java b/core/src/main/java/io/onetable/iceberg/IcebergSchemaExtractor.java index dc0f0e68..8551df71 100644 --- a/core/src/main/java/io/onetable/iceberg/IcebergSchemaExtractor.java +++ b/core/src/main/java/io/onetable/iceberg/IcebergSchemaExtractor.java @@ -67,7 +67,15 @@ public Schema toIceberg(OneSchema oneSchema) { AtomicInteger fieldIdTracker = new AtomicInteger(0); List nestedFields = convertFields(oneSchema, fieldIdTracker); List recordKeyFields = oneSchema.getRecordKeyFields(); - if (recordKeyFields.isEmpty()) { + boolean recordKeyFieldsAreNotRequired = + recordKeyFields.stream().anyMatch(f -> f.getSchema().isNullable()); + // Iceberg requires the identifier fields to be required fields, so if any of the record key + // fields are nullable, we cannot add the identifier fields to the schema properties. + if (!recordKeyFields.isEmpty() && recordKeyFieldsAreNotRequired) { + log.warn( + "Record key fields are not required. Not setting record key fields in iceberg schema."); + } + if (recordKeyFields.isEmpty() || recordKeyFieldsAreNotRequired) { return new Schema(nestedFields); } // Find field in iceberg schema that matches each of the record key path and collect ids.