RE: Using POJOs with the table API

[Alexis Sarda-Espinosa <al...@microfocus.com>]

Hello,

I've found a ticket that talks about very high-level improvements to the Table API [1]. Are there any more concrete pointers for migration from DataSet to Table API? Will it be possible at all to use POJOs with the Table API?

[1] https://issues.apache.org/jira/browse/FLINK-20787

Regards,
Alexis.

From: Alexis Sarda-Espinosa <al...@microfocus.com>
Sent: Donnerstag, 5. August 2021 15:49
To: user@flink.apache.org
Subject: Using POJOs with the table API

Hi everyone,

I had been using the DataSet API until now, but since that's been deprecated, I started looking into the Table API. In my DataSet job I have a lot of POJOs, all of which are even annotated with @TypeInfo and provide the corresponding factories. The Table API documentation talks about POJOs here: https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/types/#user-defined-data-types

I started with a single UDF to try it out (an AggregateFunction), but I have encountered several issues.

My accumulator class is a (serializable) POJO, but initially there were validation exceptions because the specified class is abstract. I added this to the class and got over that:

@FunctionHint(
        accumulator = DataTypeHint("RAW", bridgedTo = MyAccumulator.class)
)

Then there were exceptions about the output type. Since it's also a POJO, I thought this would help:

@FunctionHint(
        accumulator = DataTypeHint("RAW", bridgedTo = MyAccumulator::class),
        output = DataTypeHint("RAW", bridgedTo =  MyDTO.class)
)

But no luck: org.apache.flink.table.api.TableException: Unsupported conversion from data type 'RAW('com.MyDTO', '...')' (conversion class: com.MyDTO) to type information. Only data types that originated from type information fully support a reverse conversion.

I figured I would try something simpler and first return a List<String> from my AggregateFunction. But how do I define that in a DataTypeHint? I'm not sure if that's documented, but I looked through LogicalTypeParser and used:

output = DataTypeHint("ARRAY<STRING>")

But that throws an exception (see attachment): Table program cannot be compiled. This is a bug. Please file an issue.

I changed the List<String> to String[] and that finally worked.

Even getting a simple test running was difficult. I simply could not get TableEnvironment#fromValues to work with POJOs as input, and I tried many combinations of DataTypes.of(MyPojo.class)

At this point I still don't know how to return complex data structures encapsulated in POJOs from my UDF. Am I missing something very essential?

Regards,
Alexis.