[common][flink] Add support for complex types in kafka debezium avro cdc action

[AshishKhatkar]

Purpose

@umeshdangat has an outstanding PR which was updated after PR by @zhuangchong was merged. That PR: #3323

The above patch allows consuming data from avro data from kafka into paimon but it doesnt support complex avro types. This PR achieves that. The original PR zhuangchong#1 was pointing to @zhuangchong branch to clearly show the changes only relavant to supporting complex avro types.

This PR fixes the failing E2E tests for PR: #3931

Copying the note from PR: #3931

One issue is CdcSourceRecord contains Map<String, String> thus the current somewhat tedious approach is to deserialize avro complex types into json strings and then read them back from json strings rather than changing CdcSourceRecord Map<String, Object> to support value as Object. It would be a much larger change looking at the code changes needed.

I had to update the DataField to add a method for dataFieldEqualsIgnoreId which already existing in RichEventParser. For nested RowType fields this becomes necessary (coming from nested avro records) as when a DataField.type= RowType we cannot simply do equals on all data fields as they contain Id as well and it fails the equality, although there is no schema change.

@JingsongLi @zhuangchong let us know what you think.

cc: @umeshdangat

Linked issue: close xxx

Tests

API and Format

Documentation

[JingsongLi]

CC @yuzelin to take a reivew~

[JingsongLi]

Thanks @AshishKhatkar

[yuzelin]

I think this is unnecessary. We can store the data and let the readers truncate the data based on precision.
So we can just use Timestamp#fromEpochMillis(long milliseconds, int nanosOfMillisecond)

[AshishKhatkar]

I read the documentation for Timestamp#fromEpochMillis(long milliseconds, int nanosOfMillisecond) and it expects nanosOfMs to be in range [0, 999_999]. Are you saying we need to default to precision 6 here?

[yuzelin]

For example, if we want to store 1970-01-01 00:00:00.123456789, the first argument milliseconds is 123 and the second argument nanosOfMillisecond is 456789.

[AshishKhatkar]

@yuzelin done.

[yuzelin]

Thanks~

[yuzelin]

Thanks，@AshishKhatkar! I left a comment, please take a look~

[yuzelin]

+1