[FLINK-9964] Add a CSV table format factory

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.8.0
• Component/s: Table SQL / API
• Labels: pull-request-available

Release Note

This release introduces a new format descriptor for CSV files that is compliant
with RFC 4180. The new descriptor is available as
`org.apache.flink.table.descriptors.Csv`. For now, this can only be used
together with the Kafka connector. The old descriptor is availabla as
`org.apache.flink.table.descriptors.OldCsv` for use with file system connectors.

• Assignee: Jiayi Liao
• Reporter: Timo Walther
• Created: 26/Jul/18 07:37
• Updated: 26/Feb/19 10:56
• Resolved: 26/Feb/19 10:56

Description

We should add a RFC 4180 compliant CSV table format factory to read and write data into Kafka and other connectors. This requires a SerializationSchemaFactory and DeserializationSchemaFactory. How we want to represent all data types and nested types is still up for discussion. For example, we could flatten and deflatten nested types as it is done here. We can also have a look how tools such as the Avro to CSV tool perform the conversion.

Issue Links

• links to: GitHub Pull Request #6541
• links to: GitHub Pull Request #7777