[Connector] Pojo To InternalRow(Fluss) Utility

[MehulBatra]

Purpose
Linked issue: #723 , #569

Utility for PojoToRowData Converter for the end users to use inbuild for a quick startup.

Brief change log:

• Added utility to convert PojoToRowData
• Added tests for primitive types (boolean, numeric, floating point)
• Added tests for complex types (decimal, temporal, binary, char)
• Added edge case tests for null values and nullable fields
• Used Types.POJO(pojoClass) to get PojoTypeInfo
• Implemented the Converter Pattern, added FieldConverter interface, and Type-specific converter implementations for each data type.
• Made sure to not support for Nested Fields, as currently doesn't support one.

Tests:

• Added unit tests for all Fluss data types including boolean, numeric, decimal, date/time, and binary types, with appropriate value verification.
• Added a unit test to support that if a user passes a nested field, it would throw an error.

API and Format
No API or storage format changes.

Documentation
No documentation changes are required.

[MehulBatra]

I have covered the complex types for the conversion also.
@polyzos @wuchong Please help me with the review.

[wuchong]

Thanks for the contribution @MehulBatra , I left some comments about the implementation, please let me know if you have any questions.

[MehulBatra]

Thanks for the contribution @MehulBatra , I left some comments about the implementation, please let me know if you have any questions.

Thanks for the feedback. I will address these comments over the coming weekend and get back to you in case I am stuck!

[MehulBatra]

@wuchong Jark made changes as per the comments, please have a look and let me know if I underdid or overdid anything, I will accommodate that.

[wuchong]

@MehulBatra

The pull request looks good to me overall, but I have one key concern: it currently implements a conversion from POJO to Flink's RowData. In my opinion, what we actually need is a utility for converting POJOs to Fluss' InternalRow. This utility would be essential for writing a DataStream<POJO> to the Fluss Sink, as all data types must eventually be converted into Fluss' InternalRow—this is the data type expected by AppendWriter and UpsertWriter.

On the other hand, the conversion from POJO to Flink's RowData is already implemented by Flink itself. This is used in scenarios where a DataStream<T> is converted into a Flink Table (e.g., via org.apache.flink.table.api.bridge.java.StreamTableEnvironment#fromDataStream(org.apache.flink.streaming.api.datastream.DataStream<T>)).

To address this, I have updated the pull request to focus on converting POJOs to Fluss' InternalRow. Additionally, I have added a pre-check for the field types of the POJO to ensure compatibility and prevent potential issues.

[wuchong]

Merging...

[MehulBatra]

@MehulBatra

The pull request looks good to me overall, but I have one key concern: it currently implements a conversion from POJO to Flink's RowData. In my opinion, what we actually need is a utility for converting POJOs to Fluss' InternalRow. This utility would be essential for writing a DataStream<POJO> to the Fluss Sink, as all data types must eventually be converted into Fluss' InternalRow—this is the data type expected by AppendWriter and UpsertWriter.

On the other hand, the conversion from POJO to Flink's RowData is already implemented by Flink itself. This is used in scenarios where a DataStream<T> is converted into a Flink Table (e.g., via org.apache.flink.table.api.bridge.java.StreamTableEnvironment#fromDataStream(org.apache.flink.streaming.api.datastream.DataStream<T>)).

To address this, I have updated the pull request to focus on converting POJOs to Fluss' InternalRow. Additionally, I have added a pre-check for the field types of the POJO to ensure compatibility and prevent potential issues.

Thanks @wuchong for identifying and fixing this critical issue! I initially misunderstood the relationship between Flink's RowData and Fluss' InternalRow.
I'll make sure to pay closer attention to these format distinctions in future PRs. Really appreciate your help and guidance!