[jira] [Commented] (FLINK-34826) flink-connector-mysql-cdc-2.4.2 when a multi-column primary key index starts with a varchar column, a snapshot chunk can potentially return a large number of rows and cause jvm to OOM.

["Flink CDC Issue Import (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-34826?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17828769#comment-17828769 ] 

Flink CDC Issue Import commented on FLINK-34826:
------------------------------------------------

Date: Thu Dec 28 11:59:37 CST 2023, Author: [link3280|https://github.com/link3280]

The same issue here. @trikker Do you have some workaround that could share? Thanks!

> flink-connector-mysql-cdc-2.4.2 when a multi-column primary key index starts with a varchar column, a snapshot chunk can potentially return a large number of rows and cause jvm to OOM.
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-34826
>                 URL: https://issues.apache.org/jira/browse/FLINK-34826
>             Project: Flink
>          Issue Type: Improvement
>          Components: Flink CDC
>            Reporter: Flink CDC Issue Import
>            Priority: Major
>              Labels: github-import
>
> ### Search before asking
> - [X] I searched in the [issues|https://github.com/ververica/flink-cdc-connectors/issues) and found nothing similar.
> ### Flink version
> 1.17.1
> ### Flink CDC version
> flink-connector-mysql-cdc-2.4.2
> ### Database and its version
> source: MySQL 8.0.19
> destination: Doris 2.0.2
> ### Minimal reproduce step
> 1. create a MySQL table with multi-column primary key and its first column is varchar
> ```
> CREATE DATABASE `mydb`;
> CREATE TABLE `t` (
>   `col1` varchar(64) NOT NULL,
>   `col2` varchar(64) NOT NULL,
>   `col3` datetime(6) NOT NULL,
>   `col4` varchar(200) DEFAULT NULL,
>   `col5` varchar(500) DEFAULT NULL,
>   PRIMARY KEY (`col1`,`col2`,`col3`)
> ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
> ```
> 2. insert a lot of rows into the table
> ```
> use mydb;
> # execute the below SQL for 5000000 times
> insert into t values('aaa'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('aaa'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('aaa'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> ...
> # execute the below SQL for 5000000 times
> insert into t values('bbb'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('bbb'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('bbb'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> ...
> # execute the below SQL for 5000000 times
> insert into t values('ccc'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('ccc'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> insert into t values('ccc'), repeat('a', 64), now(6), repeat('a', 200), repeat('a', 500));
> ...
> ```
> 3. create database in doris
> ```
> CREATE DATABASE `mydb`;
> ```
> 4. create user in MySQL and doris
> In MySQL:
> ```
> create user flink identified by 'pass';
> grant all on *.* to flink;
> ```
> In Doris:
> ```
> create user flink identified by 'pass';
> grant all on *.* to flink;
> ```
> 5. download flink-doris and flink-cdc jar and put them into the lib directory of flink-1.17.1
> ```
> flink-doris-connector-1.17-1.4.0.jar
> flink-sql-connector-mysql-cdc-2.4.2.jar
> ```
> 6. use flink to sync the data from MySQL to doris
> ```
> bin/flink run -d \
>     -Dexecution.checkpointing.interval=10s \
>     -Dparallelism.default=1 \
>     -c org.apache.doris.flink.tools.cdc.CdcTools \
>     lib/flink-doris-connector-1.17-1.4.0.jar \
>     mysql-sync-database \
>     --database idc_manager \
>     --job-name flink_sync_mysql_to_doris \
>     --mysql-conf hostname=<MYSQL_IP> \
>     --mysql-conf port=3306 \
>     --mysql-conf username=flink \
>     --mysql-conf password=pass \
>     --mysql-conf database-name=mydb\
>     --including-tables "t" \
>     --sink-conf fenodes=<DORIS_FE_IP>:8030 \
>     --sink-conf username=flink \
>     --sink-conf password=pass \
>     --sink-conf jdbc-url=jdbc:mysql://<DORIS_FE_IP>:9030 \
>     --sink-conf sink.label-prefix=label1 \
>     --table-conf replication_num=3
> ```
> ### What did you expect to see?
> The table should be split into 8096 rows per chunk.
> ### What did you see instead?
> The table was split into 500M rows per chunk and finally flink job OOM.
> The split chunk SQL in MySQL will return 500M rows.
> ```
> SELECT * FROM `mydb`.`t` WHERE `col1` <= 'bbb' AND NOT (`col1` = 'bbb'];
> ```
> flink job OOM:
> ```
> Caused by: java.lang.OutOfMemoryError: Java heap space
> ```
> ### Anything else?
> When the below conditions are met, the above bad case will occur:
> 1. table with multi-column primary key;
> 2. the first column of the primary key is varchar;
> 3. there are a lot of rows with the same value of the first column;
> Root cause:
> The flink-mysql-cdc will use the first column to split the table into chunks and a chunk should contain 8096 rows approximately. But this is a ideal case. In a table with multi-column primary key, multiple rows can have the same first column value and it will make flink-cdc-mysql to read a lot of rows as a chunk because the begin and end of the chunk is only marked with the first column so it
> cannot just read 8096 rows. Instead it will read the next different value of the first column and this may result a lot of rows in a chunk.
> Suggest fix:
> Use the full column names of the primary key to split the table into chunks.
> I am willing to submit a PR but I am new to Java and I don't know how much time it will take.
> ### Are you willing to submit a PR?
> - [X] I'm willing to submit a PR!
> ---------------- Imported from GitHub ----------------
> Url: https://github.com/apache/flink-cdc/issues/2676
> Created by: [trikker|https://github.com/trikker]
> Labels: enhancement, 
> Created at: Thu Nov 09 21:01:34 CST 2023
> State: open



--
This message was sent by Atlassian Jira
(v8.20.10#820010)