fix: [iceberg] support avro union types in custom mapping

[rmannibucau]

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[Gezi-lzq]

1. automq/core/src/test/java/kafka/automq/table/binder/AvroRecordBinderTest.java

   Line 1160 in 18340f2

   public void testBindWithNestedOptionalRecord() {

2. automq/core/src/test/java/kafka/automq/table/binder/AvroRecordBinderTest.java

   Line 1056 in 18340f2

   public void testUnionFieldConversion() {

Hello @rmannibucau , I would like to understand the situation this PR aims to address, and what are the main differences compared to that test case.

[rmannibucau]

@Gezi-lzq you can remove the fix in the adapter and see how it fails. The diff is that existing test uses a struct which has a particular handling and my test a "primitive" which was not unwrapping the union and therefore not handling the logical type properly

[Gezi-lzq]

Good catch! @rmannibucau It seems we indeed missed handling a scenario for the timestamp type here.
When a timestamp type is part of a union type, sourceSchema.getLogicalType() fails to return a value at this location, leading to incorrect processing of the long variable.

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

Lines 97 to 100 in 18340f2

	protected Object convertTimestamp(Object sourceValue, Schema sourceSchema, Types.TimestampType targetType) {
	if (sourceValue instanceof Number) {
	long value = ((Number) sourceValue).longValue();
	LogicalType logicalType = sourceSchema.getLogicalType();

I believe the correct fix would be to add the condition Type.TypeID.TIMESTAMP.equals(icebergType.typeId()) to the logic here:

automq/core/src/main/java/kafka/automq/table/binder/RecordBinder.java

Lines 159 to 161 in 18340f2

	if (Type.TypeID.MAP.equals(icebergType.typeId()) || Type.TypeID.LIST.equals(icebergType.typeId())) {
	avroType = resolveUnionElement(avroType);
	}

[rmannibucau]

@Gezi-lzq sounds good, do you take it over or do you want me to update the pr next week? Happy to close this pr while a new release works for unions ;)? (Im afk until next week)

[Gezi-lzq]

@Gezi-lzq sounds good, do you take it over or do you want me to update the pr next week? Happy to close this pr while a new release works for unions ;)? (Im afk until next week)

Appreciate you pointing this out! Your work here definitely shows we need better test coverage for Union types.
I'm taking ownership of this to strengthen the testing groundwork first. Thanks again for the valuable input!
I'll ping you next week once I commit the changes, so you can easily review them. @rmannibucau