bryanck commented on code in PR #15615:
URL: https://github.com/apache/iceberg/pull/15615#discussion_r2940856801
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriterFactory.java:
##########
@@ -93,7 +97,47 @@ Table autoCreateTable(String tableName, SinkRecord sample) {
structType = SchemaUtils.toIcebergType(sample.valueSchema(),
config).asStructType();
}
- org.apache.iceberg.Schema schema = new
org.apache.iceberg.Schema(structType.fields());
+ // Get ID columns configuration
+ List<String> idColumns = config.tableConfig(tableName).idColumns();
+ Set<String> idColumnNames = Sets.newHashSet(idColumns);
+
+ // Ensure identifier fields are required and create the schema
+ List<NestedField> finalFields =
+ structType.fields().stream()
+ .map(
+ field -> {
+ if (idColumnNames.contains(field.name()) &&
field.isOptional()) {
+ // Convert identifier fields to required
+ return NestedField.required(
+ field.fieldId(), field.name(), field.type(),
field.doc());
+ }
+ return field;
+ })
+ .collect(Collectors.toList());
+
+ // Create initial schema to get field IDs
+ org.apache.iceberg.Schema initialSchema = new
org.apache.iceberg.Schema(finalFields);
+
+ // Map ID columns to field IDs
+ Set<Integer> identifierFieldIds = Sets.newHashSet();
+ if (!idColumns.isEmpty()) {
+ for (String idColumn : idColumns) {
+ NestedField field = initialSchema.findField(idColumn);
+ if (field == null) {
+ LOG.warn(
+ "ID column '{}' not found in schema for table {}, ignoring",
idColumn, tableName);
Review Comment:
Ignoring this could cause unexpected issues for systems relying on the ID.
An error is more appropriate IMO.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
