apache / apache/paimon

[Bug] Use flink Mysql CDC connector, paimon table update skip some records

Open
#2,232 7 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/incubator-paimon/issues) and found nothing similar.

### Paimon version

paimon-flink-1.16-0.5.0-incubating.jar

### Compute Engine

Flink 1.16

### Minimal reproduce step

Can't simply reproduce, happends in about 1% chance.

### What doesn't meet your expectations?

Use partial-update to write to paimon table will loss about 1% data. I have two source
```
CREATE TEMPORARY TABLE `scrm_business_opportunity` (
`id` BIGINT COMMENT '内部id',
`stage` TINYINT COMMENT '商机当前阶段',
`winning_amount` DECIMAL(11, 2) COMMENT '赢单金额',
`winning_date` DATE COMMENT '赢单日期',
`fail_reason` INTEGER COMMENT '丢单原因',
`fail_reason_desc` VARCHAR(64) COMMENT '丢单原因为【其它】时详细描述',
`last_fail_date` DATE COMMENT '上次丢单日期',
`owner_id` BIGINT COMMENT '商机归属人',
`create_user_id` BIGINT COMMENT '商机创建人',
`related_customers` VARCHAR(1023) COMMENT '商机对接人',
`related_users` VARCHAR(1023) COMMENT '商机协作人',
`stage_change_time` TIMESTAMP COMMENT '商机阶段的变更时间',
`abandon_reason` VARCHAR(128) COMMENT '废弃原因',
`tenant_id` BIGINT COMMENT '租户ID',
`deleted` BOOLEAN COMMENT '删除标记, 0正常,1-删除',
`create_time` TIMESTAMP COMMENT '创建时间',
`update_time` TIMESTAMP COMMENT '更新时间',
`proc_time` AS `PROCTIME`(),
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'database-name' = 'willing_base',
'table-name' = 'scrm_business_opportunity',
'server-id' = '...',
'hostname' = '...',
'port' = '3306',
'password' = '...',
'username' = '...',
'scan.startup.mode' = 'initial'
);

CREATE TEMPORARY TABLE `scrm_business_opportunity_ext` (
`id` BIGINT COMMENT '主键id',
`business_id` BIGINT COMMENT '商机id,scrm_business_opportunity.id',
`latest_progress_id` BIGINT COMMENT '商机最近跟进记录id,scrm_progress.id',
`change_flags` BIGINT COMMENT '商机异动标识',
`tenant_id` BIGINT COMMENT '租户',
`deleted` BOOLEAN COMMENT '删除标记, 0正常,1删除',
`create_time` TIMESTAMP COMMENT '创建时间',
`update_time` TIMESTAMP COMMENT '更新时间',
`proc_time` AS `PROCTIME`(),
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'database-name' = 'willing_base',
'table-name' = 'scrm_business_opportunity_ext',
'server-id' = '...',
'port' = '3306',
'hostname' = '...',
'password' = '...',
'username' = '...',
'scan.startup.mode' = 'initial'
);
```

I use multiple writers, and run a dedicated compaction job to resolve conflict.

My paimon table schema is:
```
CREATE TABLE IF NOT EXISTS `dwd_fact_scrm_business_opportunity_rt` (
`id` BIGINT COMMENT '内部id',
`stage` TINYINT COMMENT '商机当前阶段',
`winning_amount` DECIMAL(11, 2) COMMENT '赢单金额',
`winning_date` DATE COMMENT '赢单日期',
`latest_progress_id` BIGINT COMMENT '商机最近跟进记录id,scrm_progress.id',
`update_time` TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'merge-engine' = 'partial-update',
'bucket' = '2',
'write-only' = 'true',
'changelog-producer' = 'full-compaction',
'partial-update.ignore-delete' = 'true',
'sequence.field' = 'update_time'
);
```

the running sql is like
```
SET 'table.exec.sink.upsert-materialize' = 'NONE';

CREATE CATALOG `hdfs_catalog` WITH (
'type' = 'paimon',
'warehouse' = 'hdfs://HDFS44505/warehouse/flink'
);
USE CATALOG `hdfs_catalog`;

INSERT INTO `dwd_fact_scrm_business_opportunity_rt`
(SELECT a.`id`, a.`tenant_id`, a.`name`, ...
FROM `scrm_business_opportunity` AS a);

INSERT INTO `dwd_fact_scrm_business_opportunity_rt` (id, remark)
SELECT a.business_id, remark
FROM scrm_business_opportunity_ext AS a;
```

Flink sql job's cp is 3 minutes/per time and table compact job is 1 minite once.
### Anything else?

_No response_

### Are you willing to submit a PR?

- [Y ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with the partial-update table definition, especially partial-update.ignore-delete and sequence.field, then compare the two MySQL CDC INSERT jobs and their checkpoint and compaction schedules. Establish a reproducible case for the reported skipped records; done means identifying the cause and verifying that the records are retained with a regression test.

Written by the indexing model from the issue text.

Assessment

Tech stack
mysql, sql
Domain
data-engineering, databases, stream-processing
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.