[FLINK-27639] Flink JOIN uses the now() function when inserting data, resulting in data that cannot be deleted

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.4
• Fix Version/s: 1.16.0
• Component/s: Table SQL / API
• Assignee: Unassigned
• Reporter: lvycc
• Created: 16/May/22 12:05
• Updated: 25/Aug/22 07:20
• Resolved: 25/Aug/22 07:20

Description

I use the now() function as the field value when I insert data using SQL ，but I can't delete the inserted data，here is my sql：

//代码占位符
CREATE TABLE t_order (
    order_id INT,
    order_name STRING,
    product_id INT,
    user_id INT,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'ycc123',
    'database-name' = 'wby_test',
    'table-name' = 't_order'
);
CREATE TABLE t_logistics (
    logistics_id INT,
    logistics_target STRING,
    logistics_source STRING,
    logistics_time TIMESTAMP(0),
    order_id INT,
    PRIMARY KEY(logistics_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'ycc123',
    'database-name' = 'wby_test',
    'table-name' = 't_logistics'
);
CREATE TABLE t_join_sink (
    order_id INT,
    order_name STRING,
    logistics_id INT,
    logistics_target STRING,
    logistics_source STRING,
    logistics_time timestamp,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/wby_test?characterEncoding=utf8&useUnicode=true&useSSL=false&serverTimezone=Asia/Shanghai',
    'table-name' = 't_join_sink',
    'username' = 'root',
    'password' = 'ycc123'
);
INSERT INTO t_join_sink
SELECT ord.order_id,
ord.order_name,
logistics.logistics_id,
logistics.logistics_target,
logistics.logistics_source,
now()
FROM t_order AS ord
LEFT JOIN t_logistics AS logistics ON ord.order_id=logistics.order_id; 

The debug finds that SinkUpsertMaterializer causes the problem ，the result of the now() function changes when I delete the data，therefore, the delete operation is ignored

But what can I do to avoid this problem？

Issue Links

• is fixed by: FLINK-27849 Harden correctness for non-deterministic updates present in the changelog pipeline
• is related to: FLINK-27849 Harden correctness for non-deterministic updates present in the changelog pipeline