Use Iceberg generic record to replace Avro IndexedRecord and add a translation to/from data type in Flink

[waterlx]

Current implementation uses Avro as the intermediate fomat, as:
Input as Java Map, serialized to Avro IndexRecord (as the intermediate format), then call Parquet.write(Iceberg API) to write into Parquet file.

1. We need to remove that dependency and use Iceberg generic record/built-in types (maps with non-string keys, decimals, date/time types...) as the input (addressed)

• Remove AvroSerializer and its implementations (renamed to RecordSerializer as we need a serializer here to transform input format to Iceberg Record))
• Remove AvroUtils (addressed)
• Remove the setting on serializer in IcebergSinkAppender (Kept as we need a serializer here)

2. Add the translation to/from Flink native formats (like we did for InternalRow for Spark) (will be addressed later)

3. FileWriter#write() could accept convention type in addition to Iceberg generic record. See this comment. (will be addressed later, currently, the code only supports Iceberg Record as input )

[stevenzwu]

I 100% agree that we need to support Iceberg GenericRecord or Flink native formats. We probably need some generic serializer interface. We use Avro quite extensively in the streaming pipeline. It would be nice if we can avoid the format conversion and write Avro directly.

[waterlx]

@stevenzwu Thanks for the comments and sorry for my late response~~

Regarding:

We probably need some generic serializer interface.

Yes, I rename "AvroSerializer" into "RecordSerializer" as the generic serializer interface to transform from any input format to Iceberg Record

Regarding:

We use Avro quite extensively in the streaming pipeline. It would be nice if we can avoid the format conversion and write Avro directly

Got your idea. Opened a new issue as #943

[waterlx]

Ready to close this issue as Avro and IndexedRecord are removed, by the folllowing commits:

• 7ed7583
• 01f66ba