Why Flink Connector JDBC does't support LocalDateTime ?

[forideal]

Hello, my friends

      env: Flink 1.10, Blink Planner   

 table source

 CREATE TABLE josn_table ( order_id VARCHAR, event_time TIMESTAMP(3), proc_time AS PROCTIME() ) WITH (
'connector.properties.0.key' = 'bootstrap.servers',
 'connector.properties.0.value' = 'localhost:9092',
 'connector.property-version' = '1',
 'connector.startup-mode' = 'earliest-offset',
 'connector.topic' = 'raw',
 'connector.type' = 'kafka',
 'connector.version' = '0.11',
 'format.derive-schema' = 'true',
 'format.property-version' = '1',
 'format.type' = 'json',
 'update-mode' = 'append'
)

    mysql dim table 

CREATE TABLE ilms_box_d_order ( id VARCHAR, event_time TIMESTAMP(3)) WITH (
 'connector.type' = 'jdbc',
 'connector.url' = 'jdbc:mysql://localhost/mydb',
 'connector.table' = 'test',
 'connector.driver' = 'com.mysql.jdbc.Driver',
 'connector.username' = 'test',
 'connector.password' = 'test',
 'connector.property-version' = '1'
);

DML

INSERT INTO console_test SELECT
t1. event_time,
order_id
FROM
 josn_table
 LEFT JOIN ilms_box_d_order FOR SYSTEM_TIME AS OF josn_table.proc_time AS t1 ON josn_table.order_id = t1.id and  josn_table.event_time = t1.event_time;

When i exec this sql, i will get the follewing exception.

    Caused by: java.lang.ClassCastException: java.time.LocalDateTime cannot be cast to java.sql.Timestamp, field index: 1, field value: 2020-05-22T14:00.

https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java#L236

Why don't we support LocalDateTime? 

Best wishes.

forideal

 

[Jingsong Li]

Hi forideal,

Just because we don't have time to support it. We just support LocalDateTime in Flink after 1.9.

Welcome to contribute.

Best,

Jingsong Lee

On Fri, May 22, 2020 at 2:48 PM forideal <[hidden email]> wrote:

Hello, my friends

      env: Flink 1.10, Blink Planner   

 table source

 CREATE TABLE josn_table ( order_id VARCHAR, event_time TIMESTAMP(3), proc_time AS PROCTIME() ) WITH (
'connector.properties.0.key' = 'bootstrap.servers',
 'connector.properties.0.value' = 'localhost:9092',
 'connector.property-version' = '1',
 'connector.startup-mode' = 'earliest-offset',
 'connector.topic' = 'raw',
 'connector.type' = 'kafka',
 'connector.version' = '0.11',
 'format.derive-schema' = 'true',
 'format.property-version' = '1',
 'format.type' = 'json',
 'update-mode' = 'append'
)

    mysql dim table 

CREATE TABLE ilms_box_d_order ( id VARCHAR, event_time TIMESTAMP(3)) WITH (
 'connector.type' = 'jdbc',
 'connector.url' = 'jdbc:mysql://localhost/mydb',
 'connector.table' = 'test',
 'connector.driver' = 'com.mysql.jdbc.Driver',
 'connector.username' = 'test',
 'connector.password' = 'test',
 'connector.property-version' = '1'
);

DML

INSERT INTO console_test SELECT
t1. event_time,
order_id
FROM
 josn_table
 LEFT JOIN ilms_box_d_order FOR SYSTEM_TIME AS OF josn_table.proc_time AS t1 ON josn_table.order_id = t1.id and  josn_table.event_time = t1.event_time;

When i exec this sql, i will get the follewing exception.

    Caused by: java.lang.ClassCastException: java.time.LocalDateTime cannot be cast to java.sql.Timestamp, field index: 1, field value: 2020-05-22T14:00.

https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java#L236

Why don't we support LocalDateTime? 

Best wishes.

forideal

 

--

[Leonard Xu]

Hi, forideal

Thanks for your report, 

As Jingsong said, we do not support LocalDateTime in JDBC connector now, I create a JIRA[1] to track this.

Best,

Leonard Xu

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

在 2020年5月22日，22:07，Jingsong Li <[hidden email]> 写道：

Hi forideal,

Just because we don't have time to support it. We just support LocalDateTime in Flink after 1.9.

Welcome to contribute.

Best,

Jingsong Lee

On Fri, May 22, 2020 at 2:48 PM forideal <[hidden email]> wrote:

Hello, my friends

      env: Flink 1.10, Blink Planner   

 table source

 CREATE TABLE josn_table ( order_id VARCHAR, event_time TIMESTAMP(3), proc_time AS PROCTIME() ) WITH (
'connector.properties.0.key' = 'bootstrap.servers',
 'connector.properties.0.value' = 'localhost:9092',
 'connector.property-version' = '1',
 'connector.startup-mode' = 'earliest-offset',
 'connector.topic' = 'raw',
 'connector.type' = 'kafka',
 'connector.version' = '0.11',
 'format.derive-schema' = 'true',
 'format.property-version' = '1',
 'format.type' = 'json',
 'update-mode' = 'append'
)

    mysql dim table 

CREATE TABLE ilms_box_d_order ( id VARCHAR, event_time TIMESTAMP(3)) WITH (
 'connector.type' = 'jdbc',
 'connector.url' = 'jdbc:<a href="mysql://localhost/mydb'" class="">mysql://localhost/mydb',
 'connector.table' = 'test',
 'connector.driver' = 'com.mysql.jdbc.Driver',
 'connector.username' = 'test',
 'connector.password' = 'test',
 'connector.property-version' = '1'
);

DML

INSERT INTO console_test SELECT
t1. event_time,
order_id
FROM
 josn_table
 LEFT JOIN ilms_box_d_order FOR SYSTEM_TIME AS OF josn_table.proc_time AS t1 ON josn_table.order_id = t1.id and  josn_table.event_time = t1.event_time;

When i exec this sql, i will get the follewing exception.

    Caused by: java.lang.ClassCastException: java.time.LocalDateTime cannot be cast to java.sql.Timestamp, field index: 1, field value: 2020-05-22T14:00.

https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java#L236

Why don't we support LocalDateTime? 

Best wishes.

forideal

 

--