Introduce RowDeserializationSchema

[wuchong]

Search before asking

• I searched in the issues and found nothing similar.

Description

We have already introduced RowDataDeserializationSchema in #661, which converts Fluss LogRecord into a Flink RowData. However, that's mainly used as an internal API for SQL Connector. For DataStream API, it would be more convenient to have a RowDeserializationSchema which converts into Flink Row. The Flink Row provides both position-based access and name-based access, similar to Avro GenericRecord. In this way, users can easily add map transformation on the DataStream to convert the Row into user-defined POJOs. Such as:

        // the map transformation converts Flink Row into user-defined POJO.
        Long orderId = (Long) row.getField("orderId");
        Long itemId = (Long) row.getField("itemId");
        Integer amount = (Integer) row.getField("amount");
        String address = (String) row.getField("address");
        return new Order(orderId, itemId, amount, address);

See the discussion in #661 (comment)

Willingness to contribute

• I'm willing to submit a PR!

[MehulBatra]

@wuchong I would like to do it!

[polyzos]

@MehulBatra, I already have a working implementation of this from my original design because it is doing such conversions.
Although I need to extend the support to complex types