This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 23fb108e9b [FLINK-40340][mysql] Fix schema change exception when
parsing MySQL zerofill data types without unsigned keyword (#4500)
23fb108e9b is described below
commit 23fb108e9bea7f689230fa188e5f80c6eef0dbc3
Author: Jzjsnow <[email protected]>
AuthorDate: Fri Aug 21 10:24:22 2026 +0800
[FLINK-40340][mysql] Fix schema change exception when parsing MySQL
zerofill data types without unsigned keyword (#4500)
---
.../cdc/connectors/mysql/utils/MySqlTypeUtils.java | 26 ++++
.../mysql/source/MySqlFullTypesITCase.java | 71 +++++++++
.../mysql/source/MySqlMetadataAccessorITCase.java | 4 +-
.../mysql/source/MySqlPipelineITCase.java | 95 ++++++++++++
.../connectors/mysql/utils/MySqlTypeUtilsTest.java | 172 +++++++++++++++++++++
.../src/test/resources/ddl/column_type_test.sql | 27 +++-
.../test/resources/ddl/column_type_test_mysql8.sql | 27 +++-
.../connectors/mysql/schema/MySqlTypeUtils.java | 26 ++++
8 files changed, 445 insertions(+), 3 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java
index 4cc8758efc..3324f9fa61 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java
@@ -39,44 +39,57 @@ public class MySqlTypeUtils {
private static final String BOOL = "BOOL";
private static final String TINYINT = "TINYINT";
private static final String TINYINT_UNSIGNED = "TINYINT UNSIGNED";
+ private static final String TINYINT_ZEROFILL = "TINYINT ZEROFILL";
private static final String TINYINT_UNSIGNED_ZEROFILL = "TINYINT UNSIGNED
ZEROFILL";
private static final String SMALLINT = "SMALLINT";
private static final String SMALLINT_UNSIGNED = "SMALLINT UNSIGNED";
+ private static final String SMALLINT_ZEROFILL = "SMALLINT ZEROFILL";
private static final String SMALLINT_UNSIGNED_ZEROFILL = "SMALLINT
UNSIGNED ZEROFILL";
private static final String MEDIUMINT = "MEDIUMINT";
private static final String MEDIUMINT_UNSIGNED = "MEDIUMINT UNSIGNED";
+ private static final String MEDIUMINT_ZEROFILL = "MEDIUMINT ZEROFILL";
private static final String MEDIUMINT_UNSIGNED_ZEROFILL = "MEDIUMINT
UNSIGNED ZEROFILL";
private static final String INT = "INT";
private static final String INT_UNSIGNED = "INT UNSIGNED";
+ private static final String INT_ZEROFILL = "INT ZEROFILL";
private static final String INT_UNSIGNED_ZEROFILL = "INT UNSIGNED
ZEROFILL";
private static final String INTEGER = "INTEGER";
private static final String INTEGER_UNSIGNED = "INTEGER UNSIGNED";
+ private static final String INTEGER_ZEROFILL = "INTEGER ZEROFILL";
private static final String INTEGER_UNSIGNED_ZEROFILL = "INTEGER UNSIGNED
ZEROFILL";
private static final String BIGINT = "BIGINT";
private static final String SERIAL = "SERIAL";
private static final String BIGINT_UNSIGNED = "BIGINT UNSIGNED";
+ private static final String BIGINT_ZEROFILL = "BIGINT ZEROFILL";
private static final String BIGINT_UNSIGNED_ZEROFILL = "BIGINT UNSIGNED
ZEROFILL";
private static final String REAL = "REAL";
private static final String REAL_UNSIGNED = "REAL UNSIGNED";
+ private static final String REAL_ZEROFILL = "REAL ZEROFILL";
private static final String REAL_UNSIGNED_ZEROFILL = "REAL UNSIGNED
ZEROFILL";
private static final String FLOAT = "FLOAT";
private static final String FLOAT_UNSIGNED = "FLOAT UNSIGNED";
+ private static final String FLOAT_ZEROFILL = "FLOAT ZEROFILL";
private static final String FLOAT_UNSIGNED_ZEROFILL = "FLOAT UNSIGNED
ZEROFILL";
private static final String DOUBLE = "DOUBLE";
private static final String DOUBLE_UNSIGNED = "DOUBLE UNSIGNED";
+ private static final String DOUBLE_ZEROFILL = "DOUBLE ZEROFILL";
private static final String DOUBLE_UNSIGNED_ZEROFILL = "DOUBLE UNSIGNED
ZEROFILL";
private static final String DOUBLE_PRECISION = "DOUBLE PRECISION";
private static final String DOUBLE_PRECISION_UNSIGNED = "DOUBLE PRECISION
UNSIGNED";
+ private static final String DOUBLE_PRECISION_ZEROFILL = "DOUBLE PRECISION
ZEROFILL";
private static final String DOUBLE_PRECISION_UNSIGNED_ZEROFILL =
"DOUBLE PRECISION UNSIGNED ZEROFILL";
private static final String NUMERIC = "NUMERIC";
private static final String NUMERIC_UNSIGNED = "NUMERIC UNSIGNED";
+ private static final String NUMERIC_ZEROFILL = "NUMERIC ZEROFILL";
private static final String NUMERIC_UNSIGNED_ZEROFILL = "NUMERIC UNSIGNED
ZEROFILL";
private static final String FIXED = "FIXED";
private static final String FIXED_UNSIGNED = "FIXED UNSIGNED";
+ private static final String FIXED_ZEROFILL = "FIXED ZEROFILL";
private static final String FIXED_UNSIGNED_ZEROFILL = "FIXED UNSIGNED
ZEROFILL";
private static final String DECIMAL = "DECIMAL";
private static final String DECIMAL_UNSIGNED = "DECIMAL UNSIGNED";
+ private static final String DECIMAL_ZEROFILL = "DECIMAL ZEROFILL";
private static final String DECIMAL_UNSIGNED_ZEROFILL = "DECIMAL UNSIGNED
ZEROFILL";
private static final String CHAR = "CHAR";
private static final String VARCHAR = "VARCHAR";
@@ -146,30 +159,37 @@ public class MySqlTypeUtils {
? DataTypes.BOOLEAN()
: DataTypes.TINYINT();
case TINYINT_UNSIGNED:
+ case TINYINT_ZEROFILL:
case TINYINT_UNSIGNED_ZEROFILL:
case SMALLINT:
return DataTypes.SMALLINT();
case SMALLINT_UNSIGNED:
+ case SMALLINT_ZEROFILL:
case SMALLINT_UNSIGNED_ZEROFILL:
case INT:
case INTEGER:
case MEDIUMINT:
case MEDIUMINT_UNSIGNED:
+ case MEDIUMINT_ZEROFILL:
case MEDIUMINT_UNSIGNED_ZEROFILL:
case YEAR:
return DataTypes.INT();
case INT_UNSIGNED:
+ case INT_ZEROFILL:
case INT_UNSIGNED_ZEROFILL:
case INTEGER_UNSIGNED:
+ case INTEGER_ZEROFILL:
case INTEGER_UNSIGNED_ZEROFILL:
case BIGINT:
return DataTypes.BIGINT();
case BIGINT_UNSIGNED:
+ case BIGINT_ZEROFILL:
case BIGINT_UNSIGNED_ZEROFILL:
case SERIAL:
return DataTypes.DECIMAL(20, 0);
case FLOAT:
case FLOAT_UNSIGNED:
+ case FLOAT_ZEROFILL:
case FLOAT_UNSIGNED_ZEROFILL:
if (column.length() != FLOAT_LENGTH_UNSPECIFIED_FLAG) {
// For FLOAT types with length provided explicitly, treat
it like DOUBLE
@@ -179,22 +199,28 @@ public class MySqlTypeUtils {
}
case REAL:
case REAL_UNSIGNED:
+ case REAL_ZEROFILL:
case REAL_UNSIGNED_ZEROFILL:
case DOUBLE:
case DOUBLE_UNSIGNED:
+ case DOUBLE_ZEROFILL:
case DOUBLE_UNSIGNED_ZEROFILL:
case DOUBLE_PRECISION:
case DOUBLE_PRECISION_UNSIGNED:
+ case DOUBLE_PRECISION_ZEROFILL:
case DOUBLE_PRECISION_UNSIGNED_ZEROFILL:
return DataTypes.DOUBLE();
case NUMERIC:
case NUMERIC_UNSIGNED:
+ case NUMERIC_ZEROFILL:
case NUMERIC_UNSIGNED_ZEROFILL:
case FIXED:
case FIXED_UNSIGNED:
+ case FIXED_ZEROFILL:
case FIXED_UNSIGNED_ZEROFILL:
case DECIMAL:
case DECIMAL_UNSIGNED:
+ case DECIMAL_ZEROFILL:
case DECIMAL_UNSIGNED_ZEROFILL:
return column.length() <= 38
? DataTypes.DECIMAL(column.length(),
column.scale().orElse(0))
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java
index a0303f54d6..648325daee 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java
@@ -257,6 +257,77 @@ class MySqlFullTypesITCase extends MySqlSourceTestBase {
testMysqlPrecisionTypes(fullTypesMySql8Database);
}
+ @Test
+ void testMysql57ZerofillTypes() throws Throwable {
+ testZerofillTypes(fullTypesMySql57Database);
+ }
+
+ @Test
+ void testMysql8ZerofillTypes() throws Throwable {
+ testZerofillTypes(fullTypesMySql8Database);
+ }
+
+ private void testZerofillTypes(UniqueDatabase database) throws Exception {
+ database.createAndInitialize();
+ Boolean useLegacyJsonFormat = true;
+ CloseableIterator<Event> iterator =
+ env.fromSource(
+ getFlinkSourceProvider(
+ new String[]
{"zerofill_types"},
+ database,
+ useLegacyJsonFormat)
+ .getSource(),
+ WatermarkStrategy.noWatermarks(),
+ "Event-Source")
+ .executeAndCollect();
+
+ // skip CreateTableEvent
+ List<Event> snapshotResults =
+ MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator,
1).f0;
+ RecordData snapshotRecord = ((DataChangeEvent)
snapshotResults.get(0)).after();
+
+ RowType recordType =
+ RowType.of(
+ DataTypes.DECIMAL(20, 0).notNull(),
+ // SHOW CREATE TABLE may add UNSIGNED for ZEROFILL
types, so use the mapped
+ // types
+ DataTypes.SMALLINT(), // tiny_z_c -> SMALLINT
+ DataTypes.INT(), // small_z_c -> INT
+ DataTypes.INT(), // medium_z_c -> INT
+ DataTypes.BIGINT(), // int_z_c -> BIGINT
+ DataTypes.BIGINT(), // integer_z_c -> BIGINT
+ DataTypes.DECIMAL(20, 0), // big_z_c -> DECIMAL(20,0)
+ DataTypes.DOUBLE(), // real_z_c
+ DataTypes.FLOAT(), // float_z_c
+ DataTypes.DOUBLE(), // double_z_c
+ DataTypes.DOUBLE(), // double_precision_z_c
+ DataTypes.DECIMAL(8, 3), // numeric_z_c
+ DataTypes.DECIMAL(8, 3), // fixed_z_c
+ DataTypes.DECIMAL(8, 3) // decimal_z_c
+ );
+
+ Object[] expectedSnapshot =
+ new Object[] {
+ DecimalData.fromBigDecimal(new BigDecimal("1"), 20, 0),
+ (short) 1,
+ 2,
+ 3,
+ 4L,
+ 5L,
+ DecimalData.fromBigDecimal(new BigDecimal("6"), 20, 0),
+ 7.5d,
+ 8.5f,
+ 9.5d,
+ 10.5d,
+ DecimalData.fromBigDecimal(new BigDecimal("11.5"), 8, 3),
+ DecimalData.fromBigDecimal(new BigDecimal("12.5"), 8, 3),
+ DecimalData.fromBigDecimal(new BigDecimal("13.5"), 8, 3)
+ };
+
+ Assertions.assertThat(RecordDataTestUtils.recordFields(snapshotRecord,
recordType))
+ .isEqualTo(expectedSnapshot);
+ }
+
void testMysqlPrecisionTypes(UniqueDatabase database) throws Throwable {
RowType recordType =
RowType.of(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
index 59873291e1..fcbf1ffe31 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
@@ -374,7 +374,9 @@ class MySqlMetadataAccessorITCase extends
MySqlSourceTestBase {
database.createAndInitialize();
String[] tables =
- new String[] {"common_types", "time_types", "precision_types",
"json_types"};
+ new String[] {
+ "common_types", "time_types", "precision_types",
"json_types", "zerofill_types"
+ };
MySqlMetadataAccessor metadataAccessor = getMetadataAccessor(tables,
database, true);
assertThatThrownBy(metadataAccessor::listNamespaces)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
index a553036d56..5023b335e0 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -990,6 +990,101 @@ class MySqlPipelineITCase extends MySqlSourceTestBase {
?
DataTypes.BOOLEAN()
:
DataTypes.TINYINT())))));
+ // Add ZEROFILL types without UNSIGNED
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN
`z_tinyint` TINYINT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_tinyint",
DataTypes.SMALLINT())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN
`z_smallint` SMALLINT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_smallint",
DataTypes.INT())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN
`z_mediumint` MEDIUMINT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_mediumint",
DataTypes.INT())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN `z_int`
INT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("z_int",
DataTypes.BIGINT())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN `z_bigint`
BIGINT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_bigint",
DataTypes.DECIMAL(20, 0))))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN `z_float`
FLOAT ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("z_float",
DataTypes.FLOAT())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN `z_double`
DOUBLE ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_double",
DataTypes.DOUBLE())))));
+
+ statement.execute(
+ String.format(
+ "ALTER TABLE `%s`.`products` ADD COLUMN
`z_decimal` DECIMAL(8, 4) ZEROFILL NULL;",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn(
+ "z_decimal",
DataTypes.DECIMAL(8, 4))))));
+
// Drop orders table first to remove foreign key restraints
statement.execute(
String.format(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtilsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtilsTest.java
new file mode 100644
index 0000000000..9ce016fabb
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtilsTest.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.utils;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+
+import io.debezium.relational.Column;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+/** Unit tests for {@link MySqlTypeUtils}. */
+public class MySqlTypeUtilsTest {
+ private static final boolean TINY_INT_1_IS_BIT = true;
+
+ /** Build a nullable column with the given MySQL type name and default
length/scale. */
+ private static Column column(String typeName) {
+ return column(typeName, 10, 0);
+ }
+
+ /** Build a nullable column with the given MySQL type name, length and
scale. */
+ private static Column column(String typeName, int length, int scale) {
+ return Column.editor()
+ .name("test_col")
+ .type(typeName)
+ .length(length)
+ .scale(scale)
+ .optional(true)
+ .create();
+ }
+
+ private static DataType fromColumn(String typeName) {
+ return MySqlTypeUtils.fromDbzColumn(column(typeName),
TINY_INT_1_IS_BIT);
+ }
+
+ private static DataType fromColumn(String typeName, int length, int scale)
{
+ return MySqlTypeUtils.fromDbzColumn(column(typeName, length, scale),
TINY_INT_1_IS_BIT);
+ }
+
+ // ---- ZEROFILL (without UNSIGNED) should map the same as UNSIGNED
ZEROFILL ----
+
+ @Test
+ public void testTinyIntZerofill() {
+ assertThat(fromColumn("TINYINT
ZEROFILL")).isEqualTo(DataTypes.SMALLINT());
+ }
+
+ @Test
+ public void testSmallIntZerofill() {
+ assertThat(fromColumn("SMALLINT ZEROFILL")).isEqualTo(DataTypes.INT());
+ }
+
+ @Test
+ public void testMediumIntZerofill() {
+ assertThat(fromColumn("MEDIUMINT
ZEROFILL")).isEqualTo(DataTypes.INT());
+ }
+
+ @Test
+ public void testIntZerofill() {
+ // This is the type that triggered the original
UnsupportedOperationException.
+ assertThat(fromColumn("INT ZEROFILL")).isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ public void testIntegerZerofill() {
+ assertThat(fromColumn("INTEGER
ZEROFILL")).isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ public void testBigIntZerofill() {
+ assertThat(fromColumn("BIGINT
ZEROFILL")).isEqualTo(DataTypes.DECIMAL(20, 0));
+ }
+
+ @Test
+ public void testRealZerofill() {
+ assertThat(fromColumn("REAL ZEROFILL")).isEqualTo(DataTypes.DOUBLE());
+ }
+
+ @Test
+ public void testFloatZerofillUnspecifiedLength() {
+ // FLOAT without explicit length -> FLOAT
+ Column col =
+ Column.editor()
+ .name("test_col")
+ .type("FLOAT ZEROFILL")
+ .length(-1)
+ .optional(true)
+ .create();
+ assertThat(MySqlTypeUtils.fromDbzColumn(col, TINY_INT_1_IS_BIT))
+ .isEqualTo(DataTypes.FLOAT());
+ }
+
+ @Test
+ public void testFloatZerofillWithLength() {
+ // FLOAT with explicit length -> treated as DOUBLE
+ assertThat(fromColumn("FLOAT ZEROFILL", 10,
2)).isEqualTo(DataTypes.DOUBLE());
+ }
+
+ @Test
+ public void testDoubleZerofill() {
+ assertThat(fromColumn("DOUBLE
ZEROFILL")).isEqualTo(DataTypes.DOUBLE());
+ }
+
+ @Test
+ public void testDoublePrecisionZerofill() {
+ assertThat(fromColumn("DOUBLE PRECISION
ZEROFILL")).isEqualTo(DataTypes.DOUBLE());
+ }
+
+ @Test
+ public void testNumericZerofill() {
+ assertThat(fromColumn("NUMERIC ZEROFILL", 10,
0)).isEqualTo(DataTypes.DECIMAL(10, 0));
+ }
+
+ @Test
+ public void testFixedZerofill() {
+ assertThat(fromColumn("FIXED ZEROFILL", 10,
2)).isEqualTo(DataTypes.DECIMAL(10, 2));
+ }
+
+ @Test
+ public void testDecimalZerofill() {
+ assertThat(fromColumn("DECIMAL ZEROFILL", 10,
2)).isEqualTo(DataTypes.DECIMAL(10, 2));
+ }
+
+ // ---- ZEROFILL maps identically to UNSIGNED ZEROFILL ----
+
+ @Test
+ public void testIntZerofillEqualsIntUnsignedZerofill() {
+ assertThat(fromColumn("INT ZEROFILL")).isEqualTo(fromColumn("INT
UNSIGNED ZEROFILL"));
+ }
+
+ @Test
+ public void testBigIntZerofillEqualsBigIntUnsignedZerofill() {
+ assertThat(fromColumn("BIGINT ZEROFILL")).isEqualTo(fromColumn("BIGINT
UNSIGNED ZEROFILL"));
+ }
+
+ @Test
+ public void testDecimalZerofillEqualsDecimalUnsignedZerofill() {
+ assertThat(fromColumn("DECIMAL ZEROFILL", 10, 2))
+ .isEqualTo(fromColumn("DECIMAL UNSIGNED ZEROFILL", 10, 2));
+ }
+
+ // ---- Non-null column should produce NOT NULL type ----
+
+ @Test
+ public void testIntZerofillNotNull() {
+ Column col =
+ Column.editor()
+ .name("test_col")
+ .type("INT ZEROFILL")
+ .length(10)
+ .scale(0)
+ .optional(false)
+ .create();
+ assertThat(MySqlTypeUtils.fromDbzColumn(col, TINY_INT_1_IS_BIT))
+ .isEqualTo(DataTypes.BIGINT().notNull());
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test.sql
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test.sql
index 7f408d7fb0..61ed8899ef 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test.sql
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test.sql
@@ -202,4 +202,29 @@ VALUES (DEFAULT,
'{"key1":"value1","key2":"value2"}',
'[{"key1":"value1","key2":{"key2_1":"value2_1","key2_2":"value2_2"},"key3":["value3"],"key4":["value4_1","value4_2"]},{"key5":"value5"}]',
1
- );
\ No newline at end of file
+ );
+
+-- All ZEROFILL types without the UNSIGNED keyword. MySQL binlog preserves the
original DDL
+-- text, so these type names reach MySqlTypeUtils as e.g. "INT ZEROFILL"
during schema change
+-- parsing.
+CREATE TABLE zerofill_types
+(
+ id SERIAL,
+ tiny_z_c TINYINT ZEROFILL,
+ small_z_c SMALLINT ZEROFILL,
+ medium_z_c MEDIUMINT ZEROFILL,
+ int_z_c INT ZEROFILL,
+ integer_z_c INTEGER ZEROFILL,
+ big_z_c BIGINT ZEROFILL,
+ real_z_c REAL ZEROFILL,
+ float_z_c FLOAT ZEROFILL,
+ double_z_c DOUBLE ZEROFILL,
+ double_precision_z_c DOUBLE PRECISION ZEROFILL,
+ numeric_z_c NUMERIC(8, 3) ZEROFILL,
+ fixed_z_c FIXED(8, 3) ZEROFILL,
+ decimal_z_c DECIMAL(8, 3) ZEROFILL,
+ PRIMARY KEY (id)
+) DEFAULT CHARSET=utf8;
+
+INSERT INTO zerofill_types
+VALUES (DEFAULT, 1, 2, 3, 4, 5, 6, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5);
\ No newline at end of file
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test_mysql8.sql
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test_mysql8.sql
index 836296a720..e02c24dd27 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test_mysql8.sql
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test_mysql8.sql
@@ -207,4 +207,29 @@ VALUES (DEFAULT,
'{"key1":"value1","key2":"value2"}',
'[{"key1":"value1","key2":{"key2_1":"value2_1","key2_2":"value2_2"},"key3":["value3"],"key4":["value4_1","value4_2"]},{"key5":"value5"}]',
1
- );
\ No newline at end of file
+ );
+
+-- All ZEROFILL types without the UNSIGNED keyword. MySQL binlog preserves the
original DDL
+-- text, so these type names reach MySqlTypeUtils as e.g. "INT ZEROFILL"
during schema change
+-- parsing.
+CREATE TABLE zerofill_types
+(
+ id SERIAL,
+ tiny_z_c TINYINT ZEROFILL,
+ small_z_c SMALLINT ZEROFILL,
+ medium_z_c MEDIUMINT ZEROFILL,
+ int_z_c INT ZEROFILL,
+ integer_z_c INTEGER ZEROFILL,
+ big_z_c BIGINT ZEROFILL,
+ real_z_c REAL ZEROFILL,
+ float_z_c FLOAT ZEROFILL,
+ double_z_c DOUBLE ZEROFILL,
+ double_precision_z_c DOUBLE PRECISION ZEROFILL,
+ numeric_z_c NUMERIC(8, 3) ZEROFILL,
+ fixed_z_c FIXED(8, 3) ZEROFILL,
+ decimal_z_c DECIMAL(8, 3) ZEROFILL,
+ PRIMARY KEY (id)
+) DEFAULT CHARSET=utf8;
+
+INSERT INTO zerofill_types
+VALUES (DEFAULT, 1, 2, 3, 4, 5, 6, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5);
\ No newline at end of file
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/schema/MySqlTypeUtils.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/schema/MySqlTypeUtils.java
index 3a406ff9b0..960ce3a558 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/schema/MySqlTypeUtils.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/schema/MySqlTypeUtils.java
@@ -36,44 +36,57 @@ public class MySqlTypeUtils {
private static final String BOOL = "BOOL";
private static final String TINYINT = "TINYINT";
private static final String TINYINT_UNSIGNED = "TINYINT UNSIGNED";
+ private static final String TINYINT_ZEROFILL = "TINYINT ZEROFILL";
private static final String TINYINT_UNSIGNED_ZEROFILL = "TINYINT UNSIGNED
ZEROFILL";
private static final String SMALLINT = "SMALLINT";
private static final String SMALLINT_UNSIGNED = "SMALLINT UNSIGNED";
+ private static final String SMALLINT_ZEROFILL = "SMALLINT ZEROFILL";
private static final String SMALLINT_UNSIGNED_ZEROFILL = "SMALLINT
UNSIGNED ZEROFILL";
private static final String MEDIUMINT = "MEDIUMINT";
private static final String MEDIUMINT_UNSIGNED = "MEDIUMINT UNSIGNED";
+ private static final String MEDIUMINT_ZEROFILL = "MEDIUMINT ZEROFILL";
private static final String MEDIUMINT_UNSIGNED_ZEROFILL = "MEDIUMINT
UNSIGNED ZEROFILL";
private static final String INT = "INT";
private static final String INT_UNSIGNED = "INT UNSIGNED";
+ private static final String INT_ZEROFILL = "INT ZEROFILL";
private static final String INT_UNSIGNED_ZEROFILL = "INT UNSIGNED
ZEROFILL";
private static final String INTEGER = "INTEGER";
private static final String INTEGER_UNSIGNED = "INTEGER UNSIGNED";
+ private static final String INTEGER_ZEROFILL = "INTEGER ZEROFILL";
private static final String INTEGER_UNSIGNED_ZEROFILL = "INTEGER UNSIGNED
ZEROFILL";
private static final String BIGINT = "BIGINT";
private static final String SERIAL = "SERIAL";
private static final String BIGINT_UNSIGNED = "BIGINT UNSIGNED";
+ private static final String BIGINT_ZEROFILL = "BIGINT ZEROFILL";
private static final String BIGINT_UNSIGNED_ZEROFILL = "BIGINT UNSIGNED
ZEROFILL";
private static final String REAL = "REAL";
private static final String REAL_UNSIGNED = "REAL UNSIGNED";
+ private static final String REAL_ZEROFILL = "REAL ZEROFILL";
private static final String REAL_UNSIGNED_ZEROFILL = "REAL UNSIGNED
ZEROFILL";
private static final String FLOAT = "FLOAT";
private static final String FLOAT_UNSIGNED = "FLOAT UNSIGNED";
+ private static final String FLOAT_ZEROFILL = "FLOAT ZEROFILL";
private static final String FLOAT_UNSIGNED_ZEROFILL = "FLOAT UNSIGNED
ZEROFILL";
private static final String DOUBLE = "DOUBLE";
private static final String DOUBLE_UNSIGNED = "DOUBLE UNSIGNED";
+ private static final String DOUBLE_ZEROFILL = "DOUBLE ZEROFILL";
private static final String DOUBLE_UNSIGNED_ZEROFILL = "DOUBLE UNSIGNED
ZEROFILL";
private static final String DOUBLE_PRECISION = "DOUBLE PRECISION";
private static final String DOUBLE_PRECISION_UNSIGNED = "DOUBLE PRECISION
UNSIGNED";
+ private static final String DOUBLE_PRECISION_ZEROFILL = "DOUBLE PRECISION
ZEROFILL";
private static final String DOUBLE_PRECISION_UNSIGNED_ZEROFILL =
"DOUBLE PRECISION UNSIGNED ZEROFILL";
private static final String NUMERIC = "NUMERIC";
private static final String NUMERIC_UNSIGNED = "NUMERIC UNSIGNED";
+ private static final String NUMERIC_ZEROFILL = "NUMERIC ZEROFILL";
private static final String NUMERIC_UNSIGNED_ZEROFILL = "NUMERIC UNSIGNED
ZEROFILL";
private static final String FIXED = "FIXED";
private static final String FIXED_UNSIGNED = "FIXED UNSIGNED";
+ private static final String FIXED_ZEROFILL = "FIXED ZEROFILL";
private static final String FIXED_UNSIGNED_ZEROFILL = "FIXED UNSIGNED
ZEROFILL";
private static final String DECIMAL = "DECIMAL";
private static final String DECIMAL_UNSIGNED = "DECIMAL UNSIGNED";
+ private static final String DECIMAL_ZEROFILL = "DECIMAL ZEROFILL";
private static final String DECIMAL_UNSIGNED_ZEROFILL = "DECIMAL UNSIGNED
ZEROFILL";
private static final String CHAR = "CHAR";
private static final String VARCHAR = "VARCHAR";
@@ -139,10 +152,12 @@ public class MySqlTypeUtils {
? DataTypes.BOOLEAN()
: DataTypes.TINYINT();
case TINYINT_UNSIGNED:
+ case TINYINT_ZEROFILL:
case TINYINT_UNSIGNED_ZEROFILL:
case SMALLINT:
return DataTypes.SMALLINT();
case SMALLINT_UNSIGNED:
+ case SMALLINT_ZEROFILL:
case SMALLINT_UNSIGNED_ZEROFILL:
case INT:
case INTEGER:
@@ -151,38 +166,49 @@ public class MySqlTypeUtils {
return DataTypes.INT();
case INT_UNSIGNED:
case INT_UNSIGNED_ZEROFILL:
+ case INT_ZEROFILL:
case INTEGER_UNSIGNED:
+ case INTEGER_ZEROFILL:
case INTEGER_UNSIGNED_ZEROFILL:
case MEDIUMINT_UNSIGNED:
+ case MEDIUMINT_ZEROFILL:
case MEDIUMINT_UNSIGNED_ZEROFILL:
case BIGINT:
return DataTypes.BIGINT();
case BIGINT_UNSIGNED:
+ case BIGINT_ZEROFILL:
case BIGINT_UNSIGNED_ZEROFILL:
case SERIAL:
return DataTypes.DECIMAL(20, 0);
case FLOAT:
case FLOAT_UNSIGNED:
+ case FLOAT_ZEROFILL:
case FLOAT_UNSIGNED_ZEROFILL:
return DataTypes.FLOAT();
case REAL:
case REAL_UNSIGNED:
+ case REAL_ZEROFILL:
case REAL_UNSIGNED_ZEROFILL:
case DOUBLE:
case DOUBLE_UNSIGNED:
+ case DOUBLE_ZEROFILL:
case DOUBLE_UNSIGNED_ZEROFILL:
case DOUBLE_PRECISION:
case DOUBLE_PRECISION_UNSIGNED:
+ case DOUBLE_PRECISION_ZEROFILL:
case DOUBLE_PRECISION_UNSIGNED_ZEROFILL:
return DataTypes.DOUBLE();
case NUMERIC:
case NUMERIC_UNSIGNED:
+ case NUMERIC_ZEROFILL:
case NUMERIC_UNSIGNED_ZEROFILL:
case FIXED:
case FIXED_UNSIGNED:
+ case FIXED_ZEROFILL:
case FIXED_UNSIGNED_ZEROFILL:
case DECIMAL:
case DECIMAL_UNSIGNED:
+ case DECIMAL_ZEROFILL:
case DECIMAL_UNSIGNED_ZEROFILL:
return column.length() <= 38
? DataTypes.DECIMAL(column.length(),
column.scale().orElse(0))
