taoran92 opened a new issue, #10878: URL: https://github.com/apache/paimon/issues/10878
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Paimon version `master`, commit `030b38f9948217d0851d29b3583f4f2ee37fd7ed` (`2.2-SNAPSHOT`). ### Compute Engine - Flink 1.20.4, using Paimon's `mysql_sync_table` action. - Flink CDC 3.5.0. - MySQL 8.0.46 with row-based binlogging and `binlog_row_image=FULL`. - JDK 11.0.25, default charset UTF-8. ### Minimal reproduce step Synchronize a MySQL attachment table containing a PNG image into Paimon. The image is stored in a `BLOB` column and is 70 bytes long. 1. In MySQL, create the source table and insert a complete 1×1 PNG image: ```sql CREATE DATABASE cdc_demo; USE cdc_demo; CREATE TABLE binary_assets ( id BIGINT PRIMARY KEY, file_name VARCHAR(64), content BLOB ); INSERT INTO binary_assets VALUES ( 1, 'logo.png', FROM_BASE64('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==') ); SELECT id, OCTET_LENGTH(content) AS content_bytes, HEX(SUBSTRING(content, 1, 8)) AS png_signature FROM binary_assets; ``` MySQL returns: ```text id content_bytes png_signature 1 70 89504E470D0A1A0A ``` 2. Start the Paimon MySQL CDC action with the matching Paimon and MySQL connector dependencies installed in Flink. For this CLI example, first start a local Flink session cluster with `"$FLINK_HOME/bin/start-cluster.sh"` and run the SQL client on the same machine. This single-machine example assumes the test MySQL server listens on `127.0.0.1:33380`, has a nonzero `server-id`, uses UTC, and allows the test root account to connect with an empty password. Adjust the connection settings for your server. Set `FLINK_HOME` to the Flink installation and `PAIMON_ACTION_JAR` to the action JAR built from the Paimon commit above. ```bash "$FLINK_HOME/bin/flink" run -d -p 1 \ -Dexecution.checkpointing.interval=1s \ "$PAIMON_ACTION_JAR" mysql_sync_table \ --warehouse file:///tmp/paimon-cdc-demo \ --database ods \ --table binary_assets \ --mysql_conf hostname=127.0.0.1 \ --mysql_conf port=33380 \ --mysql_conf username=root \ --mysql_conf 'password=' \ --mysql_conf database-name=cdc_demo \ --mysql_conf table-name=binary_assets \ --mysql_conf server-id=5413 \ --mysql_conf server-time-zone=UTC \ --table_conf bucket=1 \ --table_conf sink.parallelism=1 ``` The action creates a Paimon primary-key table with `id` as the primary key and maps `content` to `BYTES`. 3. After the initial snapshot is committed, read the target in Flink SQL: ```sql SET 'execution.runtime-mode' = 'batch'; CREATE CATALOG paimon_demo WITH ( 'type' = 'paimon', 'warehouse' = 'file:///tmp/paimon-cdc-demo' ); SELECT id, TO_BASE64(content) AS content_base64 FROM paimon_demo.ods.binary_assets ORDER BY id; ``` The returned Base64 should equal the value used in the MySQL `INSERT`, but it differs. Reading the committed binary value gives 92 bytes, and the result can no longer be decoded as a PNG image. 4. To exercise the binlog path, insert another row in MySQL after the snapshot has completed: ```sql INSERT INTO cdc_demo.binary_assets SELECT 2, 'new-logo.png', content FROM cdc_demo.binary_assets WHERE id = 1; ``` Wait for a checkpoint and repeat the Paimon query. The new row has the same corruption. ### What doesn't meet your expectations? Binary values should be preserved byte for byte when synchronizing a MySQL `BLOB` into a Paimon `BYTES` column. Both the initial snapshot and subsequent binlog inserts commit corrupted data without reporting a conversion error: ```text Snapshot row Binlog row MySQL value 70 bytes 70 bytes Paimon value 92 bytes 92 bytes PNG decoding after readback Failed Failed ``` For example, the PNG signature changes from: ```text Expected: 89 50 4E 47 0D 0A 1A 0A Actual: EF BF BD 50 4E 47 0D 0A 1A 0A ``` This makes binary attachments unreadable even though the synchronization job continues to commit records. ### Anything else? I reproduced this with a real MySQL server and `MySqlSyncTableAction` in an in-process Flink MiniCluster, then read the checkpoint-committed Paimon table and attempted to decode the stored image. The command above is the equivalent CLI invocation of the action. The captured Debezium events declare `content` as plain `bytes`. Their Base64 payload decodes to the original 70 bytes on both the snapshot and binlog paths, so the source event still contains the correct image. In `DebeziumSchemaUtils.transformRawValue`, the ordinary `bytes` branch converts the decoded binary value to a Java string: ```java // Arbitrary binary bytes are interpreted using the default charset. transformed = new String(Base64.getDecoder().decode(rawValue)); ``` The sink later converts the intermediate string back to UTF-8 bytes for `VARBINARY`. Bytes such as the PNG signature's `0x89` have already been replaced with `U+FFFD` and cannot be recovered by re-encoding. This occurs with the default charset already set to UTF-8. A local change that preserves Base64 with explicit binary-field metadata until sink conversion keeps both records at 70 bytes, byte-identical to the source, and the images decode successfully. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
