fix(binder): enhance RecordBinder and TypeAdapter to support STRUCT type conversion

[Gezi-lzq]

This pull request refactors and extends the Avro-to-Iceberg record binding and conversion logic, improving support for struct, list, and map types while simplifying field mapping and adapter interfaces.

fix #3009
fixed protobuf timestamp type

[Copilot]

Pull Request Overview

This pull request refactors the Avro-to-Iceberg record binding and type conversion logic to better support nested struct types within lists and maps. The key changes include:

• Introduced a new StructConverter functional interface to enable recursive struct conversion
• Extended TypeAdapter interface with an additional method accepting a StructConverter parameter
• Refactored RecordBinder to pre-compute binders for nested structs in lists and maps
• Extracted FieldMapping into a separate public class for better code organization
• Simplified field mapping initialization logic in RecordBinder
• Added test coverage for lists of records and maps with record values
• Removed the writeRecordWithConvertedMapEntries helper method as the functionality is now handled by the refactored type adapters

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 11 comments.

Show a summary per file

File	Description
TypeAdapter.java	Added new convert method overload accepting StructConverter for recursive struct conversion
StructConverter.java	New functional interface for converting struct-typed values during type adaptation
AbstractTypeAdapter.java	Updated convert method to delegate STRUCT type conversion to StructConverter callback
AvroValueAdapter.java	Implemented convertStruct method and bridged old/new convert method signatures
RecordBinder.java	Refactored field mapping initialization and extended binder pre-computation to support structs in lists/maps
FieldMapping.java	Extracted as a standalone class to represent Avro-to-Iceberg field mappings
AvroRecordBinderTest.java	Added tests for list-of-records and map-with-record-values conversion; removed obsolete helper method

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

core/src/main/java/kafka/automq/table/binder/AvroValueAdapter.java:134

• The recursive call to convert in convertList should use the 4-parameter overload with the StructConverter to maintain consistency with the new callback pattern. Currently it calls the 3-parameter method which will use this::convertStruct from line 191, but this breaks encapsulation and could cause issues if the caller intended to use a different StructConverter.

            Object convert = convert(element, elementSchema, targetType.elementType());

core/src/main/java/kafka/automq/table/binder/AvroValueAdapter.java:165

• The recursive calls to convert in convertMap should use the 4-parameter overload with the StructConverter callback parameter to maintain consistency with the new callback pattern and ensure proper handling of nested structs.

                Object key = convert(record.get(keyField.pos()), keySchema, keyType);
                Object value = convert(record.get(valueField.pos()), valueSchema, valueType);

core/src/main/java/kafka/automq/table/binder/AvroValueAdapter.java:183

• The recursive calls to convert should use the 4-parameter overload with the StructConverter callback parameter to maintain consistency with the new callback pattern and ensure proper handling of nested structs in map values.

            Object key = convert(rawKey, STRING_SCHEMA_INSTANCE, keyType);
            Object value = convert(entry.getValue(), valueSchema, valueType);

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.