Hi all! I have data stored in Postgres tables where a column is of XML data type. Is Flink compatible with this data type? And which Flink data type should be used to read it? Thanks!

Its not mapped to Flinks type system, per https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/#data-type-mapping

Thanks! Is there a way to bring it in a different format and process it in the stream?

I’m not sure. I know that it’s already a problem for things like UUID so I doubt it works out of the box. It’s an open item IIRC

It seems like it can be parsed with STRING type. Not yet sure if that has other downsides.

Probably not. But not sure if that will work if you want to write XML to the Postgres database

Luckily I do not have to write XML to Postgres for my use case! 🙂

BTW see https://issues.apache.org/jira/browse/FLINK-29267 for the latter 🙂

Cool thanks! I had some other question about the types that I not get fully yet. In the context of schema evolution (see https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/fault-tolerance/serialization/schema_evolution/), it says that the only supported data types are POJO or Avro. If one uses the default Flink types would that work with schema evolution? I am not fully getting what these POJO/Avro types are and how they relate to the default types.

There's a difference between Flink's data types in the Table API/SQL landscape https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/types/ and the data types that a user of a DataStream API application can use to store their state. Since the DataStream API is a low-level API, it allows users to define their own data type when they want to store state. With the declarative way of the Table API, that's not possible, because there's a fixed set of data types that a user can define there and everything is mapped to that

And with both default Flink datatypes (Table as well as Datastream) schema_evolution is possible? For custom data types it is only possible if implemented from the POJO type?

With DataStream API, you can definitely evolve schema, depending on the serializer that you choose. With Table API, since it's declarative, you can't evolve the schema yourself. There also should not be a need for that normally. The only downside is that an upgrade of Flink with a new planner could result in a different query plan compared to the old version, which would render your snapshots as incompatible

I might not fully get it yet, but even for the Table API that would be needed right? For example, if I have a col1 for a certain sink table. And then because of business logic I would be needed to create another column (col2), which is e.g. col2=col1 +1. If I understand it correct, this would be possible using Datastream API and not using Table API? Good to know btw is that I am trying to use Flink using the CDC connector to sync my Postgres database with other databases.

If you would add another column, it would basically be a metadata change. Dynamic tables don’t materialize, so there is no state stored. There is only state stored in cases like joins, where it could be that an outputted event needs to be retracted.

Ok cool! Thanks for the help.