[FLINK-22174] `csv.quote-character` does not work for csv format when sink

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
• Assignee: Unassigned
• Reporter: hehuiyuan
• Created: 09/Apr/21 08:04
• Updated: 09/Apr/21 08:12

Description

CREATE TABLE kafkaTableSource (
    name string,
    age int,
    sex string,
    address string
) WITH (
    'connector' = 'kafka',
    'topic' = 'hehuiyuan1',
    'scan.startup.mode' = 'latest-offset',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.client.id' = 'test-consumer-group',
    'properties.group.id' = 'test-consumer-group',
    'format' = 'csv',
    'csv.quote-character' = '*'
);

CREATE TABLE kafkaTableSink (
    name string,
    age int,
    sex string,
    address string
) WITH (
    'connector' = 'kafka',
    'topic' = 'hehuiyuan2',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'csv',
    'csv.quote-character' = '#'
);

insert into kafkaTableSink select * from kafkaTableSource;

 

Test 1 :

Generate data for topic `hehuiyuan1`

>hehuiyuan,22,man,hengshui
>*hehuiyuan*,*22*,*man*,*hengshui*
>#hehuiyuan#,22,#man#,#hengshui#
>hehuiyuan,22,#man#,#hengshui#
>"hehuiyuan",22,#man#,#hengshui#

 

Result data for topic `hehuiyuan2`

>hehuiyuan,22,man,hengshui
>hehuiyuan,22,man,hengshui
>###hehuiyuan###,22,###man###,###hengshui###
>hehuiyuan,22,###man###,###hengshui###
>#"hehuiyuan"#,22,###man###,###hengshui###

I think the result shoule be ` #hehuiyuan#,22,#man#,#hengshui# ` for the first and second data.

 

Test 2:

Add Timestamp field:

CREATE TABLE kafkaTableSource (
    name string,
    age int,
    sex string,
    address string,
    dt timestamp(3)
) WITH (

Generate data for topic `hehuiyuan1`

>hehuiyuan,22,man,hengshui,2020-12-12 12:12:12
>hehuiyuan,22,man,hengshui,2020-12-16 10:00:00

Result data for topic `hehuiyuan2`

>hehuiyuan,22,man,hengshui,#2020-12-12 12:12:12#
>hehuiyuan,22,man,hengshui,#2020-12-16 10:00:00#