twthorn commented on code in PR #16826:
URL: https://github.com/apache/iceberg/pull/16826#discussion_r3449223773
##########
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:
Good to know, updated to reflect this.
##########
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:
Updated as well
##########
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:
Good catch, updated to align, and reflected this in tests as well.
--
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]
