[Bug] Use flink Mysql CDC connector, paimon table update skip some records
- 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