laskoviymishka commented on code in PR #16826:
URL: https://github.com/apache/iceberg/pull/16826#discussion_r3449074931
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -254,11 +258,19 @@ private GenericRecord convertToStruct(
hasSchemaUpdates = true;
}
}
+ Object recordFieldValue = struct.get(recordField);
+ if (recordFieldValue == null && schemaUpdateConsumer != null) {
Review Comment:
Small one: this fires the null-walk even when `hasSchemaUpdates` is already
true for the field, so when the field itself needs an updateType/makeOptional
we end up emitting at this level twice. It's harmless today because
`SchemaUpdate.Consumer` dedups by name, but it's leaning on that implicitly.
Adding `&& !hasSchemaUpdates` here keeps this consistent with the non-null
branch (which defers child discovery to the next record) and drops the
redundant walk.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java:
##########
@@ -268,6 +280,89 @@ private GenericRecord convertToStruct(
return result;
}
+ /**
+ * Recursively traverses the Connect schema and emits all evolution events
(addColumn, updateType,
+ * and makeOptional) at every nested level.
+ *
+ * <p>Unlike {@link #convertToStruct(Struct, StructType, int,
SchemaUpdate.Consumer)} which skips
+ * a field's children once an update is detected (deferring nested discovery
to re-conversion),
+ * this method always recurses through the full schema.
+ */
+ private void evolveSchemaFromConnectSchema(
+ org.apache.kafka.connect.data.Schema recordSchema,
+ Type tableType,
+ int tableFieldId,
+ SchemaUpdate.Consumer schemaUpdateConsumer) {
+ if (recordSchema == null) {
+ return;
+ }
+ switch (recordSchema.type()) {
+ case STRUCT:
+ if (tableType.isStructType()) {
+ StructType structType = tableType.asStructType();
+ for (Field field : recordSchema.fields()) {
+ NestedField nestedField = lookupStructField(field.name(),
structType, tableFieldId);
+ if (nestedField == null) {
+ String parentFieldName =
+ tableFieldId < 0 ? null :
tableSchema.findColumnName(tableFieldId);
+ Type type = SchemaUtils.toIcebergType(field.schema(), config);
+ schemaUpdateConsumer.addColumn(parentFieldName, field.name(),
type);
+ } else {
+ PrimitiveType evolveDataType =
+ SchemaUtils.needsDataTypeUpdate(nestedField.type(),
field.schema());
+ if (evolveDataType != null) {
+ String fieldName =
tableSchema.findColumnName(nestedField.fieldId());
+ schemaUpdateConsumer.updateType(fieldName, evolveDataType);
+ }
+ if (nestedField.isRequired() && field.schema().isOptional()) {
+ String fieldName =
tableSchema.findColumnName(nestedField.fieldId());
+ schemaUpdateConsumer.makeOptional(fieldName);
+ }
+ evolveSchemaFromConnectSchema(
+ field.schema(), nestedField.type(), nestedField.fieldId(),
schemaUpdateConsumer);
+ }
+ }
+ } else {
+ logMismatchedType(recordSchema.type(), tableType);
+ }
+ break;
+ case ARRAY:
+ if (tableType.isListType()) {
+ ListType listType = tableType.asListType();
+ evolveSchemaFromConnectSchema(
+ recordSchema.valueSchema(),
+ listType.elementType(),
+ listType.elementId(),
+ schemaUpdateConsumer);
+ } else {
+ logMismatchedType(recordSchema.type(), tableType);
+ }
+ break;
+ case MAP:
+ if (tableType.isMapType()) {
+ MapType mapType = tableType.asMapType();
+ evolveSchemaFromConnectSchema(
+ recordSchema.keySchema(), mapType.keyType(), mapType.keyId(),
schemaUpdateConsumer);
Review Comment:
This is the one thing I'd fix before merge, and it's a small change.
Recursing into `keySchema()`/`keyId()` here emits evolution events against the
map *key* struct, but Iceberg treats map keys as immutable —
`SchemaUpdate.commit()` rejects add/update/makeOptional on anything inside a
key struct ("Cannot add fields to map keys" / "Cannot update map keys" /
"Cannot alter map keys").
So a null map field whose Connect key struct gained a field emits
`addColumn(parent="...key", ...)`, and the next `IcebergWriter` flush runs
`applySchemaUpdates` → `commit()`:
```
addColumn(parent="data.key", name="k2", StringType)
-> SchemaUpdate.commit()
-> map(): parentToAddedIds.containsKey(keyId)
-> throw IllegalArgumentException("Cannot add fields to map keys")
```
which takes the record (and the task) down.
`testNestedSchemaEvolutionMapKeyWithNullValue` doesn't catch it because it
inspects the consumer directly and never calls `applySchemaUpdates`.
I'd just recurse into the value subtree only (drop the key recursion) — that
mirrors how `SchemaUpdate.internalAddColumn` already redirects map column adds
to the value type — and flip that test to assert we emit nothing for the key
(or to expect the throw through `applySchemaUpdates`). The rest of the
recursion looks great.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordConverter.java:
##########
@@ -859,6 +859,411 @@ private void assertTypesAddedFromStruct(Function<String,
Type> fn) {
assertThat(fn.apply("ma")).isInstanceOf(MapType.class);
}
+ @Test
+ public void testNestedSchemaEvolutionStructWithNullValue() {
+ org.apache.iceberg.Schema nestedStructSchema =
+ new org.apache.iceberg.Schema(
+ NestedField.required(1, "id", IntegerType.get()),
+ NestedField.optional(
+ 2, "nested", StructType.of(NestedField.required(3, "a",
IntegerType.get()))));
+
+ Table table = mock(Table.class);
+ when(table.schema()).thenReturn(nestedStructSchema);
+ RecordConverter converter = new RecordConverter(table, config);
+
+ Schema connectNestedSchema =
+ SchemaBuilder.struct()
+ .optional()
+ .field("a", Schema.INT32_SCHEMA)
+ .field("b", Schema.OPTIONAL_STRING_SCHEMA)
+ .build();
+ Schema connectSchema =
+ SchemaBuilder.struct()
+ .field("id", Schema.INT32_SCHEMA)
+ .field("nested", connectNestedSchema)
+ .build();
+ Struct data = new Struct(connectSchema).put("id", 1).put("nested", null);
+
+ SchemaUpdate.Consumer consumer = new SchemaUpdate.Consumer();
+ Record result = converter.convert(data, consumer);
+
+ assertThat(result.getField("id")).isEqualTo(1);
+ assertThat(result.getField("nested")).isNull();
+
+ Collection<AddColumn> addCols = consumer.addColumns();
+ assertThat(addCols).hasSize(1);
+ AddColumn addCol = addCols.iterator().next();
+ assertThat(addCol.parentName()).isEqualTo("nested");
+ assertThat(addCol.name()).isEqualTo("b");
+ assertThat(addCol.type()).isInstanceOf(StringType.class);
+ }
+
+ @Test
+ public void testNoSchemaEvolutionStructWithNullValue() {
+ org.apache.iceberg.Schema nestedStructSchema =
+ new org.apache.iceberg.Schema(
+ NestedField.required(1, "id", IntegerType.get()),
+ NestedField.optional(
+ 2, "nested", StructType.of(NestedField.required(3, "a",
IntegerType.get()))));
+
+ Table table = mock(Table.class);
+ when(table.schema()).thenReturn(nestedStructSchema);
+ RecordConverter converter = new RecordConverter(table, config);
+
+ Schema connectNestedSchema =
+ SchemaBuilder.struct().optional().field("a",
Schema.INT32_SCHEMA).build();
+ Schema connectSchema =
+ SchemaBuilder.struct()
+ .field("id", Schema.INT32_SCHEMA)
+ .field("nested", connectNestedSchema)
+ .build();
+ Struct data = new Struct(connectSchema).put("id", 1).put("nested", null);
+
+ SchemaUpdate.Consumer consumer = new SchemaUpdate.Consumer();
+ Record result = converter.convert(data, consumer);
+
+ assertThat(result.getField("id")).isEqualTo(1);
+ assertThat(result.getField("nested")).isNull();
+
+ assertThat(consumer.addColumns()).isEmpty();
+ assertThat(consumer.makeOptionals()).isEmpty();
+ assertThat(consumer.updateTypes()).isEmpty();
+ assertThat(consumer.empty()).isTrue();
+ }
+
+ @Test
+ public void testNestedSchemaEvolutionListOfStructsWithNullValue() {
+ org.apache.iceberg.Schema tableSchema =
+ new org.apache.iceberg.Schema(
+ NestedField.required(
+ 1,
+ "items",
+ ListType.ofRequired(
+ 2,
+ StructType.of(
+ NestedField.required(3, "product_id",
IntegerType.get()),
+ NestedField.optional(
+ 4,
+ "details",
+ StructType.of(NestedField.required(5, "name",
StringType.get())))))));
+
+ Table table = mock(Table.class);
+ when(table.schema()).thenReturn(tableSchema);
+ RecordConverter converter = new RecordConverter(table, config);
+
+ Schema detailsSchema =
+ SchemaBuilder.struct()
+ .optional()
+ .field("name", Schema.OPTIONAL_STRING_SCHEMA)
+ .field("price", Schema.OPTIONAL_FLOAT64_SCHEMA)
+ .build();
+ Schema itemSchema =
+ SchemaBuilder.struct()
+ .field("product_id", Schema.INT32_SCHEMA)
+ .field("details", detailsSchema)
+ .build();
+ Schema connectSchema =
+ SchemaBuilder.struct().field("items",
SchemaBuilder.array(itemSchema).build()).build();
+
+ Struct item = new Struct(itemSchema).put("product_id", 101).put("details",
null);
+ Struct data = new Struct(connectSchema).put("items",
ImmutableList.of(item));
+
+ SchemaUpdate.Consumer consumer = new SchemaUpdate.Consumer();
+ Record result = converter.convert(data, consumer);
+
+ List<Record> items = (List<Record>) result.getField("items");
Review Comment:
These unchecked casts (this one and the `(Map<String, Record>)` cast in
`testNestedSchemaEvolutionMapOfStructsWithNullValue`) will warn under
`-Xlint:unchecked` and can fail CI where warnings are errors. The existing
`testMapToString`/`testStructToString` in this file already carry
`@SuppressWarnings("unchecked")` — adding the same annotation to these two
keeps the build clean.
--
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]
