You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "annurahar (via GitHub)" <gi...@apache.org> on 2026/02/21 18:47:06 UTC
[PR] KafkaConnect: Fix Nested schema evolution [iceberg]
annurahar opened a new pull request, #15396:
URL: https://github.com/apache/iceberg/pull/15396
Closes #15395
## Summary
This PR fixes a bug where the Kafka Connect sink does not evolve schema for nested fields inside `struct<struct<...>>`, `list<struct<...>>`, or `map<string, struct<...>>` when schema evolution is enabled.
## Problem
When `iceberg.tables.evolve-schema-enabled=true`, the `RecordConverter` only checks top-level fields for schema evolution. It does not recursively traverse nested fields inside complex types. As a result:
- New nested fields defined in the record schema are silently ignored
- The Iceberg table schema is never updated to include these nested fields
- No error or warning is thrown — the connector continues without issue
### Example Scenario
**Record Schema** (Kafka Connect):
```
orders (
order_id: int
customer: struct
- customer_id: int
- customer_details: struct (NEW) <-- Not detected
- name: string
- email: string
)
```
**Iceberg Table Schema** (before):
```
orders (
order_id: int
customer: struct
- customer_id: int
)
```
**Expected Behavior**: The `customer_details` nested struct should be added to the Iceberg table schema.
**Actual Behavior**: The `customer_details` field is ignored; schema remains unchanged.
## Solution
### Key Changes
1. **Added `evolveSchema(SinkRecord, SchemaUpdate.Consumer)` method** in `RecordConverter`:
- Separate public method for schema evolution
- Traverses the entire record schema to detect all schema changes (new fields, type updates, optionality changes)
- Does NOT convert data - only collects schema updates
2. **Simplified `convert(SinkRecord)` method**:
- Now only handles data conversion
- No longer takes `SchemaUpdate.Consumer` parameter
3. **Added recursive schema evolution** for nested fields:
- `evolveSchemaFromStruct()` - Handles Struct records with schema
- `evolveSchemaFromMap()` - Handles schemaless Map records
- `evolveFieldSchema()` - Common logic for processing individual fields
- `evolveNestedSchemaFromStruct()` - Recursively processes nested fields in Struct records
- `evolveNestedSchemaFromMapValue()` - Recursively processes nested fields in Map records
- `evolveMapAsStruct()` - Processes Map values as struct fields
- `evolveFirstStructElementInList()` - Handles schema evolution for `list<struct>`
- `evolveFirstStructValueInMap()` - Handles schema evolution for `map<string, struct>`
4. **Correctly determines parent field names** for nested columns:
- `struct<struct>` → uses parent struct's field ID
- `list<struct>` → uses list element's field ID
- `map<string, struct>` → uses map value's field ID
## Configuration
To enable nested schema evolution, the connector must be configured to fetch and use the most recent schema version from the Schema Registry, rather than relying on the schema embedded in each message. Add the following configuration properties:
```properties
# Disable embedded schema in messages
value.converter.schemas.enable=false
key.converter.schemas.enable=false
# Enable enhanced Avro schema support
value.converter.enhanced.avro.schema.support=true
key.converter.enhanced.avro.schema.support=true
# Use latest schema version from Schema Registry
key.converter.use.latest.version=true
value.converter.use.latest.version=true
# Enable auto schema evolution
value.converter.auto.evolve=true
key.converter.auto.evolve=true
```
## Testing
Added two new test cases:
1. **`testNestedSchemaEvolutionFromStruct`** - Tests nested schema evolution using Struct records (with schema)
2. **`testNestedSchemaEvolutionFromMap`** - Tests nested schema evolution using Map records (schemaless)
Both tests verify that:
- New fields inside nested structs are detected
- Correct parent field name is set (e.g., `outer.inner`)
- Correct field type is inferred
All existing tests continue to pass.
## Checklist
- [x] New fields inside `struct<struct<...>>` are detected and added
- [x] New fields inside `list<struct<...>>` are detected and added
- [x] New fields inside `map<string, struct<...>>` are detected and added
- [x] Type updates for nested primitive fields are detected
- [x] Optionality changes for nested fields are detected
- [x] Works for both Struct (with schema) and Map (schemaless) records
- [x] No breaking changes to existing functionality
- [x] All tests pass
- [x] Code formatted with Spotless
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#issuecomment-4686364813
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#discussion_r2940889399
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java:
##########
@@ -126,8 +128,20 @@ private static void commitSchemaUpdates(Table table, SchemaUpdate.Consumer updat
update -> updateSchema.addColumn(update.parentName(), update.name(), update.type()));
updateTypes.forEach(update -> updateSchema.updateColumn(update.name(), update.type()));
makeOptionals.forEach(update -> updateSchema.makeColumnOptional(update.name()));
- updateSchema.commit();
- LOG.info("Schema for table {} updated with new columns", table.name());
+ try {
+ updateSchema.commit();
+ LOG.info("Schema for table {} updated with new columns", table.name());
+ } catch (ValidationException | CommitFailedException e) {
+ // this can happen if the field already exists in the table, which is fine (other task
+ // committed the schema). No action required
Review Comment:
These exceptions could be caused for other reasons as well, and thus we could lose data if we ignore the schema update in those cases.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#issuecomment-4627188135
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on code in PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#discussion_r3032556075
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java:
##########
@@ -126,8 +128,20 @@ private static void commitSchemaUpdates(Table table, SchemaUpdate.Consumer updat
update -> updateSchema.addColumn(update.parentName(), update.name(), update.type()));
updateTypes.forEach(update -> updateSchema.updateColumn(update.name(), update.type()));
makeOptionals.forEach(update -> updateSchema.makeColumnOptional(update.name()));
- updateSchema.commit();
- LOG.info("Schema for table {} updated with new columns", table.name());
+ try {
+ updateSchema.commit();
+ LOG.info("Schema for table {} updated with new columns", table.name());
+ } catch (ValidationException | CommitFailedException e) {
+ // this can happen if the field already exists in the table, which is fine (other task
+ // committed the schema). No action required
Review Comment:
Yes this exception catching is dangerous. Also to handle the scenario where other task has added the new column we have
```
Tasks.range(1)
.retry(IcebergSinkConfig.SCHEMA_UPDATE_RETRIES)
.run(notUsed -> commitSchemaUpdates(table, updates));
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#discussion_r2989654675
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java:
##########
@@ -126,8 +128,20 @@ private static void commitSchemaUpdates(Table table, SchemaUpdate.Consumer updat
update -> updateSchema.addColumn(update.parentName(), update.name(), update.type()));
updateTypes.forEach(update -> updateSchema.updateColumn(update.name(), update.type()));
makeOptionals.forEach(update -> updateSchema.makeColumnOptional(update.name()));
- updateSchema.commit();
- LOG.info("Schema for table {} updated with new columns", table.name());
+ try {
+ updateSchema.commit();
+ LOG.info("Schema for table {} updated with new columns", table.name());
+ } catch (ValidationException | CommitFailedException e) {
+ // this can happen if the field already exists in the table, which is fine (other task
+ // committed the schema). No action required
Review Comment:
If the commit fails and the column is not added, then wouldn't we lose data in that field when we write the data?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "kzajaczkowski (via GitHub)" <gi...@apache.org>.
kzajaczkowski commented on PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#issuecomment-4379277258
@kumarpritam863 the argument that existing code already traverses nested structs only holds when **the parent value is non-null at every level**. The current `convertStructValue` does:
```java
for (Field recordField : recordSchema.fields()) {
Object value = struct.get(recordField);
if (value == null) {
continue; // null parent → no recursion → no addColumn
}
// schema-update path...
}
```
When a parent struct value is null in the current record but has new sub-fields in the *schema*, those sub-fields are never detected, even though the typed schema is sitting right there on `record.valueSchema()`.
A practical case where this matters is a Protobuf `oneof` whose arms started as zero-field marker messages and were later evolved to carry data. Generic example:
**V1**:
```protobuf
message OrderLifecycleEvent {
string order_id = 1;
oneof event {
OrderPlacedEvent placed = 2;
OrderShippedEvent shipped = 3;
OrderCancelledEvent cancelled = 4;
}
}
message OrderPlacedEvent {}
message OrderShippedEvent {}
message OrderCancelledEvent {}
```
The Iceberg sink with `auto-create-enabled: true` creates the table with `event.placed: struct<>`, `event.shipped: struct<>`, `event.cancelled: struct<>`.
**V2** adds a Timestamp to each arm (BACKWARD-compatible per Schema Registry):
```protobuf
message OrderPlacedEvent { google.protobuf.Timestamp placed_at = 1; }
message OrderShippedEvent { google.protobuf.Timestamp shipped_at = 1; }
message OrderCancelledEvent { google.protobuf.Timestamp cancelled_at = 1; }
```
Confluent's ProtobufConverter wraps the oneof in a synthetic outer struct where exactly one arm is non-null per record and the other two are null. With value-driven recursion on a V2 record that sets only `shipped`:
- `shipped` → non-null → recurse → `addColumn(shipped_at)` fires ✓
- `placed` → null → `continue` → no `addColumn(placed_at)` ✗
- `cancelled` → null → `continue` → no `addColumn(cancelled_at)` ✗
Only one arm gets evolved per record. The other two remain `struct<>` in the table — possibly forever, if a producer emits a homogeneous stream that never sets a particular arm.
This PR's schema-driven `evolveSchema` (iterating `recordField.schema().fields()` regardless of value) is exactly the right design for this case — it discovers all three arms' new fields from a single record. The value-driven defense above doesn't cover null sibling fields.
One thing worth considering for the empty-struct variant: even with detection working correctly, the call order in `IcebergWriter.convertToRow` (`flush()` before `applySchemaUpdates`) still tries to close the existing writer that was created with the V1 empty-struct table schema. `BaseTaskWriter.complete()` → `ParquetWriter.close()` → `ensureWriterInitialized()` → `new ParquetFileWriter(parquetSchema)` → `TypeUtil.checkValidWriteSchema()` rejects empty groups, so the inline flush throws `InvalidSchemaException` before `applySchemaUpdates` runs. The new try/catch around that block would swallow the exception silently and `initNewWriter()` would recreate against the same un-evolved schema, looping. Might be worth either skipping empty-struct columns at auto-create (in `SchemaUtils.toIcebergType` STRUCT case) or applying schema updates before flush in the empty-existing-struct case. Happy to file or contribute either as a follow-up.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15396: KafkaConnect: Fix Nested schema evolution
URL: https://github.com/apache/iceberg/pull/15396
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#issuecomment-4367589700
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "annurahar (via GitHub)" <gi...@apache.org>.
annurahar commented on code in PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#discussion_r2986442100
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/SchemaUtils.java:
##########
@@ -126,8 +128,20 @@ private static void commitSchemaUpdates(Table table, SchemaUpdate.Consumer updat
update -> updateSchema.addColumn(update.parentName(), update.name(), update.type()));
updateTypes.forEach(update -> updateSchema.updateColumn(update.name(), update.type()));
makeOptionals.forEach(update -> updateSchema.makeColumnOptional(update.name()));
- updateSchema.commit();
- LOG.info("Schema for table {} updated with new columns", table.name());
+ try {
+ updateSchema.commit();
+ LOG.info("Schema for table {} updated with new columns", table.name());
+ } catch (ValidationException | CommitFailedException e) {
+ // this can happen if the field already exists in the table, which is fine (other task
+ // committed the schema). No action required
Review Comment:
data loss will never occur as exception is not thrown. It is gracefully handled. In the finally block here https://github.com/apache/iceberg/pull/15396/changes#diff-e4d2ba65bde87ccf54b9537c5a14c79151b4d6db742c31c72b0a04fd5367f36dR111 new instance of table writer is created which is used further to write data to the table
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] KafkaConnect: Fix Nested schema evolution [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #15396:
URL: https://github.com/apache/iceberg/pull/15396#issuecomment-4183124967
@annurahar
the current code actually traverses to the nested schema if you observe carefully. For the case you have described
```
orders (
order_id: int
customer: struct
- customer_id: int
- customer_details: struct (NEW) <-- Not detected
- name: string
- email: string
)
```
When customer_details are added then while traversing it will first look for order_id and it will find it in the tables and will convert value from connect to iceberg and move to customer.
When coming to customer
```
NestedField tableField = lookupStructField(recordField.name(), schema, structFieldId);
```
This will find the Nested Field as customer does exist and it will proceed with the Customer for convertValue but as customer being a struct, if you see the convertValue method
```
private Object convertValue(
Object value, Type type, int fieldId, SchemaUpdate.Consumer schemaUpdateConsumer) {
if (value == null) {
return null;
}
switch (type.typeId()) {
case STRUCT:
return convertStructValue(value, type.asStructType(), fieldId, schemaUpdateConsumer);
case LIST:
return convertListValue(value, type.asListType(), schemaUpdateConsumer);
case MAP:
return convertMapValue(value, type.asMapType(), schemaUpdateConsumer);
case INTEGER:
return convertInt(value);
case LONG:
return convertLong(value);
case FLOAT:
return convertFloat(value);
case DOUBLE:
return convertDouble(value);
case DECIMAL:
return convertDecimal(value, (DecimalType) type);
case BOOLEAN:
return convertBoolean(value);
case STRING:
return convertString(value);
case UUID:
return convertUUID(value);
case BINARY:
return convertBase64Binary(value);
case FIXED:
return ByteBuffers.toByteArray(convertBase64Binary(value));
case DATE:
return convertDateValue(value);
case TIME:
return convertTimeValue(value);
case TIMESTAMP:
return convertTimestampValue(value, (TimestampType) type);
}
throw new UnsupportedOperationException("Unsupported type: " + type.typeId());
}
```
It again calls "convertStructValue" with the schema of struct (which here will be the schema of the "customer" and with "customer" as the parent.
Now the convertStructValue will iterate through the "customer" and will first find "customer_id" which is in the table so just convert the value and for "customer_details" the NestedField will be null and this block get's activated:
```
if (tableField == null) {
// add the column if schema evolution is on, otherwise skip the value
if (schemaUpdateConsumer != null) {
String parentFieldName =
structFieldId < 0 ? null : tableSchema.findColumnName(structFieldId);
Type type = SchemaUtils.toIcebergType(recordField.schema(), config);
schemaUpdateConsumer.addColumn(parentFieldName, recordField.name(), type);
}
}
```
This is a very primitive nesting, we run kafka-connect-iceberg at huge scales with schema referring other schemas and even with the referred schema being evolved our table gets evolved to the new state.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org