diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlDefaultValueConverter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlDefaultValueConverter.java index 4f6869633d3..227eafb2352 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlDefaultValueConverter.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlDefaultValueConverter.java @@ -60,7 +60,7 @@ public class MySqlDefaultValueConverter implements DefaultValueConverter { Pattern.compile("([0-9]*-[0-9]*-[0-9]*) ([0-9]*:[0-9]*:[0-9]*(\\.([0-9]*))?)"); private static final Pattern CHARSET_INTRODUCER_PATTERN = - Pattern.compile("^_[A-Za-z0-9]+'(.*)'$"); + Pattern.compile("^_[A-Za-z0-9]+(?:'(.*)'|\"(.*)\")$"); @Immutable private static final Set TRIM_DATA_TYPES = @@ -161,6 +161,16 @@ public Object convert(Column column, String value) { // strip character set introducer on default value expressions value = stripCharacterSetIntroducer(value); + // A default value like `_utf8mb4' 0 '` will become ` 0 ` after stripping the character + // set introducer, i.e. paddings inside the quotes still remain. For numeric / date-like + // columns those inner paddings will crash the downstream parsers (NumberFormatException, + // DateTimeParseException, etc.). Trim once more here so every downstream `convertToXxx` + // gets a clean literal. VARCHAR-like columns keep their original value since they are + // filtered out by TRIM_DATA_TYPES. + if (TRIM_DATA_TYPES.contains(column.jdbcType())) { + value = value.trim(); + } + // boolean is also INT(1) or TINYINT(1) if (NUMBER_DATA_TYPES.contains(column.jdbcType()) && ("true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value))) { @@ -522,6 +532,12 @@ private String replaceFirstNonNumericSubstring(String s, int startIndex, char c) private String stripCharacterSetIntroducer(String value) { final Matcher matcher = CHARSET_INTRODUCER_PATTERN.matcher(value); - return !matcher.matches() ? value : matcher.group(1); + if (!matcher.matches()) { + return value; + } + // group(1) matches when the literal is wrapped by single quotes, + // group(2) matches when it is wrapped by double quotes. Only one of them is non-null. + final String singleQuoted = matcher.group(1); + return singleQuoted != null ? singleQuoted : matcher.group(2); } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlConnectorITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlConnectorITCase.java index 6a44906b27d..dc1f04638af 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlConnectorITCase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlConnectorITCase.java @@ -1621,6 +1621,265 @@ void testAlterWithDefaultStringValue() throws Exception { jobClient.cancel().get(); } + /** + * Integration test for ALTER statements whose default value looks like {@code _utf8mb4' 0 '}. + * + *

After being rewritten by the MySQL optimizer, such default values combine a character-set + * introducer with padding spaces inside the quotes, e.g. {@code ALTER TABLE t ADD COLUMN c + * TINYINT(4) DEFAULT _UTF8MB4' 0 ' COMMENT '...'}. Once the DDL is shipped through the binlog, + * {@code MySqlDefaultValueConverter} must parse it correctly, otherwise a NumberFormatException + * will fail the job. + * + *

Assertion mechanism: because {@link RestartStrategyUtils#configureNoRestartStrategy} + * disables restarts, a single failing DDL will terminate the job immediately, so the expected + * DELETE event can never be produced and the case fails. Therefore this case serves as both a + * positive verification (all DDLs are parsed successfully and the DELETE event is consumed) and + * a reverse reproduction (removing either the trim or the introducer-stripping logic will make + * the case fail). + */ + @Test + void testAlterWithCharsetIntroducerDefaultValue() throws Exception { + setup(true); + RestartStrategyUtils.configureNoRestartStrategy(env); + customerDatabase.createAndInitialize(); + String sourceDDL = + String.format( + "CREATE TABLE default_value_test (" + + " id BIGINT NOT NULL," + + " name STRING," + + " address STRING," + + " phone_number BIGINT," + + " primary key (id) not enforced" + + ") WITH (" + + " 'connector' = 'mysql-cdc'," + + " 'hostname' = '%s'," + + " 'port' = '%s'," + + " 'username' = '%s'," + + " 'password' = '%s'," + + " 'database-name' = '%s'," + + " 'table-name' = '%s'," + + " 'scan.incremental.snapshot.enabled' = 'true'," + + " 'server-time-zone' = 'UTC'," + + " 'server-id' = '%s'," + + " 'scan.incremental.snapshot.chunk.size' = '%s'" + + ")", + MYSQL_CONTAINER.getHost(), + MYSQL_CONTAINER.getDatabasePort(), + customerDatabase.getUsername(), + customerDatabase.getPassword(), + customerDatabase.getDatabaseName(), + "default_value_test", + getServerId(true), + getSplitSize(true)); + tEnv.executeSql(sourceDDL); + // async submit job + TableResult result = tEnv.executeSql("SELECT * FROM default_value_test"); + JobClient jobClient = result.getJobClient().get(); + waitForJobStatus( + jobClient, + Collections.singletonList(RUNNING), + Deadline.fromNow(Duration.ofSeconds(10))); + CloseableIterator iterator = result.collect(); + waitForSnapshotStarted(iterator); + // Wait 1s until snapshot phase finished; DDL cannot be applied during snapshot phase. + List actualRows = new ArrayList<>(fetchRows(iterator, 2)); + Thread.sleep(1000L); + + String[] expected = + new String[] { + "+I[1, user1, Shanghai, 123567]", + "+I[2, user2, Shanghai, 123567]", + "-D[1, user1, Shanghai, 123567]" + }; + + // Ship a batch of DDLs with character-set-introducer default values via the binlog first, + // then trigger the DELETE. Ordering requirement (key to the reverse-reproduction ability): + // the DELETE must happen AFTER every DDL, so that once MySqlDefaultValueConverter fails to + // parse the padding inside the quotes, the SourceReader will be stuck and the -D event + // will never be consumed; `fetchRows` then fails the case due to the no-restart strategy + // combined with the missing event. + try (Connection connection = customerDatabase.getJdbcConnection(); + Statement statement = connection.createStatement()) { + // 1) introducer + no padding inside quotes -- baseline case + statement.execute( + "alter table default_value_test add column `c_tiny_plain` TINYINT(4) DEFAULT _utf8mb4'0';"); + // 2) introducer + padding inside quotes -- the core case of this fix + // (requires the second trim after stripCharacterSetIntroducer) + statement.execute( + "alter table default_value_test add column `c_tiny_inner_pad` TINYINT(4) DEFAULT _utf8mb4' 0 ' COMMENT 'directive business';"); + // 3) mixed-case introducer + padding inside quotes -- verifies the regex is + // case-insensitive + statement.execute( + "alter table default_value_test add column `c_int_upper` INT DEFAULT _UTF8MB4' 1 ';"); + // 4) INT + introducer + padding inside quotes -- reproduces the 42 case from the + // original question + statement.execute( + "alter table default_value_test add column `c_int_answer` INT DEFAULT _utf8mb4' 42 ';"); + // 5) BIGINT + introducer + padding inside quotes + statement.execute( + "alter table default_value_test add column `c_bigint` BIGINT DEFAULT _utf8mb4' 10086 ';"); + // 6) DECIMAL + introducer + padding inside quotes + statement.execute( + "alter table default_value_test add column `c_decimal` DECIMAL(8,2) DEFAULT _utf8mb4' 3.14 ';"); + // 7) DATE + introducer + padding inside quotes -- covers the DATE branch + statement.execute( + "alter table default_value_test add column `c_date` DATE DEFAULT _utf8mb4' 2024-01-02 ';"); + // 8) DATETIME + introducer + padding inside quotes -- covers the TIMESTAMP branch + statement.execute( + "alter table default_value_test add column `c_dt` DATETIME DEFAULT _utf8mb4' 2024-01-02 03:04:05 ';"); + // 9) TINYINT(1) + introducer -- covers the boolean type going through + // NUMBER_DATA_TYPES + statement.execute( + "alter table default_value_test add column `c_bool` TINYINT(1) DEFAULT _utf8mb4'1';"); + // 10) double-quoted literal + padding inside quotes -- directly reproduces the + // `_UTF8MB4" 0 "` case from the original question + statement.execute( + "alter table default_value_test add column `c_dq_inner_pad` TINYINT(4) DEFAULT _UTF8MB4\" 0 \" COMMENT 'directive business';"); + // 11) double-quoted literal + DECIMAL -- covers the DECIMAL branch parsing a + // double-quoted value + statement.execute( + "alter table default_value_test add column `c_dq_decimal` DECIMAL(8,2) DEFAULT _utf8mb4\" 3.14 \";"); + // Trigger the DELETE only after every DDL has succeeded, so the -D event can only + // appear after the schema evolution completes normally. + statement.execute("DELETE FROM default_value_test WHERE id=1;"); + } + actualRows.addAll(fetchRows(iterator, expected.length - 2)); + assertEqualsInAnyOrder(Arrays.asList(expected), actualRows); + jobClient.cancel().get(); + } + + /** + * Integration test for the snapshot code path when a table is created with columns whose + * {@code DEFAULT} clause carries a character-set introducer plus padding spaces inside the + * quotes (both single-quoted {@code _utf8mb4' 0 '} and double-quoted {@code _UTF8MB4" 0 "} + * forms). + * + *

Default values are parsed in two different code paths inside the connector: during the + * snapshot phase the schema is discovered by {@code SHOW CREATE TABLE} whose output is + * re-parsed by the Debezium Antlr parser, and during the incremental phase the {@code ALTER + * TABLE} statements coming from the binlog are parsed on the fly. Both eventually flow through + * {@link io.debezium.connector.mysql.MySqlDefaultValueConverter#convert}. + * + *

Empirically the MySQL server normalizes such default value expressions on the + * server side before persisting them, so {@code SHOW CREATE TABLE} returns the fully + * canonicalized form (e.g. {@code DEFAULT '0'}) regardless of what the user originally wrote. + * That means the snapshot path never actually observes the introducer / padding shape and + * cannot serve as a reverse reproduction for this bug (the reverse reproduction is exclusively + * covered by {@link #testAlterWithCharsetIntroducerDefaultValue()} which feeds the raw DDL to + * Debezium via the binlog). This case therefore acts as a positive verification that the + * current fix does not introduce any side effect on the snapshot schema discovery path: the + * source must reach RUNNING, the snapshot must yield the initial rows, and a subsequent binlog + * INSERT relying on server-side defaults must also flow through, wiring both parsing paths end + * to end. Combined with the no-restart strategy, any regression on the snapshot side would fail + * the case via {@code fetchRows} timing out. + */ + @Test + void testSnapshotWithCharsetIntroducerDefaultValue() throws Exception { + setup(true); + RestartStrategyUtils.configureNoRestartStrategy(env); + customerDatabase.createAndInitialize(); + + // Create a dedicated table with character-set-introducer defaults BEFORE the Flink source + // starts, so the whole schema (including these defaults) is discovered through + // `SHOW CREATE TABLE` during the snapshot phase, i.e. the exact code path the review + // comment asks us to cover. + try (Connection connection = customerDatabase.getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE snapshot_cs_default_test (" + + " id INT NOT NULL PRIMARY KEY," + // single-quoted introducer, no padding -- baseline + + " c_tiny_plain TINYINT(4) DEFAULT _utf8mb4'0'," + // single-quoted introducer, padding inside quotes -- core case + + " c_tiny_inner_pad TINYINT(4) DEFAULT _utf8mb4' 0 '," + // upper-case introducer, padding inside quotes + + " c_int_upper INT DEFAULT _UTF8MB4' 1 '," + // INT with padding inside quotes + + " c_int_answer INT DEFAULT _utf8mb4' 42 '," + // BIGINT with padding inside quotes + + " c_bigint BIGINT DEFAULT _utf8mb4' 10086 '," + // DECIMAL with padding inside quotes + + " c_decimal DECIMAL(8,2) DEFAULT _utf8mb4' 3.14 '," + // DATE with padding inside quotes -- exercises the DATE branch + + " c_date DATE DEFAULT _utf8mb4' 2024-01-02 '," + // DATETIME with padding inside quotes -- exercises the TIMESTAMP branch + + " c_dt DATETIME DEFAULT _utf8mb4' 2024-01-02 03:04:05 '," + // TINYINT(1) -- boolean going through NUMBER_DATA_TYPES + + " c_bool TINYINT(1) DEFAULT _utf8mb4'1'," + // double-quoted introducer, padding inside quotes -- directly + // reproduces the `_UTF8MB4\" 0 \"` shape from the original bug report + + " c_dq_inner_pad TINYINT(4) DEFAULT _UTF8MB4\" 0 \"," + // double-quoted introducer + DECIMAL branch + + " c_dq_decimal DECIMAL(8,2) DEFAULT _utf8mb4\" 3.14 \"" + + ");"); + // Two initial rows -- will be produced by the snapshot phase. + statement.execute("INSERT INTO snapshot_cs_default_test (id) VALUES (1),(2);"); + } + + // Declare Flink table with a subset of columns; we don't need to project every default + // column, we only need the source to build the schema successfully -- which forces every + // default value to be parsed. Reading `id` alongside two default-driven columns is enough + // to observe that (a) the schema parsed OK and (b) the values MySQL filled via those + // defaults survived the end-to-end path. + String sourceDDL = + String.format( + "CREATE TABLE snapshot_cs_default_test (" + + " id INT NOT NULL," + + " c_tiny_inner_pad TINYINT," + + " c_dq_inner_pad TINYINT," + + " primary key (id) not enforced" + + ") WITH (" + + " 'connector' = 'mysql-cdc'," + + " 'hostname' = '%s'," + + " 'port' = '%s'," + + " 'username' = '%s'," + + " 'password' = '%s'," + + " 'database-name' = '%s'," + + " 'table-name' = '%s'," + + " 'scan.incremental.snapshot.enabled' = 'true'," + + " 'server-time-zone' = 'UTC'," + + " 'server-id' = '%s'," + + " 'scan.incremental.snapshot.chunk.size' = '%s'" + + ")", + MYSQL_CONTAINER.getHost(), + MYSQL_CONTAINER.getDatabasePort(), + customerDatabase.getUsername(), + customerDatabase.getPassword(), + customerDatabase.getDatabaseName(), + "snapshot_cs_default_test", + getServerId(true), + getSplitSize(true)); + tEnv.executeSql(sourceDDL); + + TableResult result = tEnv.executeSql("SELECT * FROM snapshot_cs_default_test"); + JobClient jobClient = result.getJobClient().get(); + // If any of the character-set-introducer defaults blows up during snapshot schema + // discovery the job never reaches RUNNING and this assertion fails fast. + waitForJobStatus( + jobClient, + Collections.singletonList(RUNNING), + Deadline.fromNow(Duration.ofSeconds(10))); + CloseableIterator iterator = result.collect(); + waitForSnapshotStarted(iterator); + + // Snapshot phase must produce the two rows inserted before source startup. + List actualRows = new ArrayList<>(fetchRows(iterator, 2)); + Thread.sleep(1000L); + + // Insert another row through the binlog, relying on server-side defaults for every column + // other than `id`. This exercises the incremental path as well and, together with the + // snapshot rows, ties the whole two-path story together. + try (Connection connection = customerDatabase.getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute("INSERT INTO snapshot_cs_default_test (id) VALUES (3);"); + } + + String[] expected = new String[] {"+I[1, 0, 0]", "+I[2, 0, 0]", "+I[3, 0, 0]"}; + actualRows.addAll(fetchRows(iterator, expected.length - 2)); + assertEqualsInAnyOrder(Arrays.asList(expected), actualRows); + jobClient.cancel().get(); + } + @ParameterizedTest(name = "incrementalSnapshot = {0}") @ValueSource(booleans = {true, false}) void testStartupFromSpecificBinlogFilePos(boolean incrementalSnapshot) throws Exception {