[FLINK-22163] DataTypes implementation for flink-csv with ZonedDateTime support

• Type: Sub-task
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.12.1
• Component/s: Table SQL / Ecosystem
• Assignee: Unassigned
• Reporter: François Lacombe
• Created: 08/Apr/21 17:26
• Updated: 14/Apr/21 12:00

Description

Dear maintainers

flink-csv's CsvRowSerializationSchema currently rely on legacy TypeInformation to serialize java Types.
It doesn't support ZonedDateTime (only LocalDateTime with LOCAL_DATE_TIME)

An exception is thrown when CsvRowSerializationSchema is built with a RowTypeInformaton containing a TypeInformation.of(ZonedDateTime.class). See CsvRowSchemaConverter.java line 201.

java.lang.IllegalArgumentException: Unsupported type information 'GenericType<java.time.ZonedDateTime>' for field 'periode'.
 at org.apache.flink.formats.csv.CsvRowSchemaConverter.convertType(CsvRowSchemaConverter.java:201)
 at org.apache.flink.formats.csv.CsvRowSchemaConverter.convert(CsvRowSchemaConverter.java:155)
 at org.apache.flink.formats.csv.CsvRowSerializationSchema$Builder.<init>(CsvRowSerializationSchema.java:111)
 ...

A quick fix would be to use `toString()` by default in the serializer instead of throwing an Exception. Apache CSV already does this way.

 

I'm using flink-csv and Apache Flink 1.12.1

Am I missing anything regarding a more recent flink-csv implementation?

If not, is it planned to move flink-csv to DataTypes types (with DataTypes.TIMESTAMP_WITH_TIME_ZONE support)?

 

All the best