Yanquan Lv created FLINK-41485:
----------------------------------
Summary: [mysql][pipeline] Column reordering is not propagated,
causing silent field misalignment
Key: FLINK-41485
URL: https://issues.apache.org/jira/browse/FLINK-41485
Project: Flink
Issue Type: Bug
Components: Flink CDC
Affects Versions: cdc-3.6.0, cdc-3.5.0, cdc-3.4.0
Reporter: Yanquan Lv
h2. Description
The MySQL CDC Pipeline does not propagate column order changes introduced by
the following DDL statements:
* MODIFY COLUMN ... FIRST / AFTER ...
* CHANGE COLUMN ... FIRST / AFTER ...
After such a change, MySQL/Debezium uses the updated column order for
subsequent records, while downstream operators still interpret those records
using the previous schema.
When the reordered columns have the same or compatible data types, the job may
continue running without errors while producing incorrect field values.
h2. Environment
* Flink: <VERSION>
* Flink CDC: <AFFECTED_VERSION_WITHOUT_THE_FIX>
* MySQL: 8.0, with Binlog enabled, binlog_format=ROW, and binlog_row_image=FULL.
* Kafka: An empty topic named cdc-column-reorder-repro, preferably with one
partition.
* Matching MySQL and Kafka Pipeline Connector dependencies installed.
* A MySQL CDC account with the required snapshot and replication permissions.
* Start a fresh job without restoring from a checkpoint or savepoint.
h2. Steps to Reproduce
h3. 1. Prepare the source table
Execute the following statements in MySQL:
{code:sql}
CREATE DATABASE cdc_repro;
USE cdc_repro;
CREATE TABLE column_reorder (
id INT NOT NULL,
name VARCHAR(64),
city VARCHAR(64),
PRIMARY KEY (id)
);
INSERT INTO column_reorder (id, name, city)
VALUES (1, 'Alice', 'Hangzhou');
{code}
The initial column order is id, name, city.
Both name and city use the same data type to demonstrate silent field
misalignment rather than a type conversion failure.
h3. 2. Configure and submit the pipeline
Save the following configuration as column-reorder.yaml. Replace the connection
placeholders and ensure the server-id range does not overlap with other
replication clients.
{code:yaml}
source:
type: mysql
hostname: <MYSQL_HOST>
port: 3306
username: <CDC_USER>
password: <CDC_PASSWORD>
tables: cdc_repro.column_reorder
server-id: 5400-5404
server-time-zone: UTC
scan.startup.mode: initial
sink:
type: kafka
properties.bootstrap.servers: <KAFKA_HOST>:9092
topic: cdc-column-reorder-repro
value.format: debezium-json
pipeline:
name: mysql-column-reorder-repro
parallelism: 1
schema.change.behavior: evolve
{code}
Submit the job from the CDC distribution directory:
{code:bash}
./bin/flink-cdc.sh column-reorder.yaml
{code}
h3. 3. Verify the initial snapshot
Consume the Kafka topic:
{code:bash}
<KAFKA_HOME>/bin/kafka-console-consumer.sh \
--bootstrap-server <KAFKA_HOST>:9092 \
--topic cdc-column-reorder-repro \
--from-beginning
{code}
Wait until the initial snapshot record is received. Its after object should
contain:
{code:json}
{"id":1,"name":"Alice","city":"Hangzhou"}
{code}
Only the after object is shown here, not the complete Debezium JSON message.
h3. 4. Reorder an existing column
Execute the following statements after the initial record has been synchronized:
{code:sql}
USE cdc_repro;
ALTER TABLE column_reorder
MODIFY COLUMN city VARCHAR(64) AFTER id;
SHOW COLUMNS FROM column_reorder;
{code}
Verify that the source column order is now id, city, name.
h3. 5. Insert a new record
{code:sql}
INSERT INTO column_reorder (id, name, city)
VALUES (2, 'Bob', 'Shanghai');
SELECT id, name, city
FROM column_reorder
WHERE id = 2;
{code}
The MySQL query returns id=2, name='Bob', and city='Shanghai'.
Inspect the Kafka message for id=2 and compare values by field name, regardless
of JSON property order.
h2. Actual Result
On affected versions, downstream operators interpret records using the previous
column order, id, name, city, although the incoming values follow the updated
order, id, city, name.
The resulting after object is:
{code:json}
{"id":2,"name":"Shanghai","city":"Bob"}
{code}
The name and city values are swapped. Since both columns are VARCHAR(64), the
job may remain running without reporting an error.
h2. Expected Result
Column order changes should be correctly recognized and handled throughout the
pipeline. Subsequent records must preserve the association between column names
and values.
The after object for id=2 should contain:
{code:json}
{"id":2,"name":"Bob","city":"Shanghai"}
{code}
Sinks that cannot physically reorder columns should preserve correct field
mappings or fail explicitly according to the schema evolution policy, rather
than silently writing misaligned values.
h2. Root Cause
Before the fix, CustomAlterTableParserListener emits type change events for
MODIFY COLUMN and type change/rename events for CHANGE COLUMN, but does not
emit an event representing the FIRST/AFTER column movement.
Consequently, the source record layout and the downstream schema can become
inconsistent.
h2. Acceptance Criteria
* Correctly handle FIRST/AFTER clauses in MODIFY COLUMN and CHANGE COLUMN.
* Preserve correct field values for inserts, updates, and deletes after column
reordering.
* Cover simultaneous column renaming and reordering.
* Cover multiple column changes within a single ALTER TABLE statement.
* Prevent silent field misalignment in sinks that do not support physical
column reordering.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
