You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ru...@apache.org on 2024/11/11 07:09:13 UTC
(flink-cdc) branch master updated: [FLINK-36408][pipeline-connector][mysql] Fix MySQL pipeline connector failed to parse FLOAT type with precision (#3625)
This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 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 57a1a7df9 [FLINK-36408][pipeline-connector][mysql] Fix MySQL pipeline connector failed to parse FLOAT type with precision (#3625)
57a1a7df9 is described below
commit 57a1a7df93a69dbbd37fe4e3e5e4146cbe1f9983
Author: yuxiqian <34...@users.noreply.github.com>
AuthorDate: Mon Nov 11 15:09:08 2024 +0800
[FLINK-36408][pipeline-connector][mysql] Fix MySQL pipeline connector failed to parse FLOAT type with precision (#3625)
---
.../docs/connectors/pipeline-connectors/mysql.md | 5 +-
.../docs/connectors/pipeline-connectors/mysql.md | 5 +-
.../cdc/connectors/mysql/utils/MySqlTypeUtils.java | 10 +-
.../mysql/source/MySqlFullTypesITCase.java | 41 +++++-
.../mysql/source/MySqlMetadataAccessorITCase.java | 150 +++++++++++++++++++++
.../src/test/resources/ddl/column_type_test.sql | 26 +++-
.../test/resources/ddl/column_type_test_mysql8.sql | 32 ++++-
7 files changed, 257 insertions(+), 12 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
index 4d5edb66f..a5f0924c3 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
@@ -393,7 +393,10 @@ source:
DOUBLE UNSIGNED ZEROFILL<br>
DOUBLE PRECISION<br>
DOUBLE PRECISION UNSIGNED<br>
- DOUBLE PRECISION UNSIGNED ZEROFILL
+ DOUBLE PRECISION UNSIGNED ZEROFILL<br>
+ FLOAT(p, s)<br>
+ REAL(p, s)<br>
+ DOUBLE(p, s)
</td>
<td>DOUBLE</td>
<td></td>
diff --git a/docs/content/docs/connectors/pipeline-connectors/mysql.md b/docs/content/docs/connectors/pipeline-connectors/mysql.md
index 36c4cc770..3cc5c17a1 100644
--- a/docs/content/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content/docs/connectors/pipeline-connectors/mysql.md
@@ -403,7 +403,10 @@ source:
DOUBLE UNSIGNED ZEROFILL<br>
DOUBLE PRECISION<br>
DOUBLE PRECISION UNSIGNED<br>
- DOUBLE PRECISION UNSIGNED ZEROFILL
+ DOUBLE PRECISION UNSIGNED ZEROFILL<br>
+ FLOAT(p, s)<br>
+ REAL(p, s)<br>
+ DOUBLE(p, s)
</td>
<td>DOUBLE</td>
<td></td>
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 0e70ed6d9..c82525cf6 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
@@ -107,6 +107,7 @@ public class MySqlTypeUtils {
private static final String MULTIPOLYGON = "MULTIPOLYGON";
private static final String MULTILINESTRING = "MULTILINESTRING";
private static final String UNKNOWN = "UNKNOWN";
+ private static final int FLOAT_LENGTH_UNSPECIFIED_FLAG = -1;
/** Returns a corresponding Flink data type from a debezium {@link Column}. */
public static DataType fromDbzColumn(Column column) {
@@ -164,7 +165,12 @@ public class MySqlTypeUtils {
case FLOAT:
case FLOAT_UNSIGNED:
case FLOAT_UNSIGNED_ZEROFILL:
- return DataTypes.FLOAT();
+ if (column.length() != FLOAT_LENGTH_UNSPECIFIED_FLAG) {
+ // For FLOAT types with length provided explicitly, treat it like DOUBLE
+ return DataTypes.DOUBLE();
+ } else {
+ return DataTypes.FLOAT();
+ }
case REAL:
case REAL_UNSIGNED:
case REAL_UNSIGNED_ZEROFILL:
@@ -236,7 +242,7 @@ public class MySqlTypeUtils {
return DataTypes.ARRAY(DataTypes.STRING());
default:
throw new UnsupportedOperationException(
- String.format("Don't support MySQL type '%s' yet.", typeName));
+ String.format("MySQL type '%s' is not supported yet.", typeName));
}
}
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 2bd1ae689..dd15fdf6a 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
@@ -249,7 +249,18 @@ public class MySqlFullTypesITCase extends MySqlSourceTestBase {
DataTypes.TIMESTAMP_LTZ(0),
DataTypes.TIMESTAMP_LTZ(3),
DataTypes.TIMESTAMP_LTZ(6),
- DataTypes.TIMESTAMP_LTZ(0));
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE());
Object[] expectedSnapshot =
new Object[] {
@@ -265,7 +276,19 @@ public class MySqlFullTypesITCase extends MySqlSourceTestBase {
TimestampData.fromTimestamp(Timestamp.valueOf("2020-07-17 18:00:22")),
LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:00")),
LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
- LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22"))
+ LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
+ 2d,
+ 3d,
+ 5d,
+ 7d,
+ 11d,
+ 13d,
+ 17d,
+ 19d,
+ 23d,
+ 29d,
+ 31d,
+ 37d
};
Object[] expectedStreamRecord =
@@ -282,7 +305,19 @@ public class MySqlFullTypesITCase extends MySqlSourceTestBase {
TimestampData.fromTimestamp(Timestamp.valueOf("2020-07-17 18:00:22")),
LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:00")),
LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
- LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22"))
+ LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
+ 2d,
+ 3d,
+ 5d,
+ 7d,
+ 11d,
+ 13d,
+ 17d,
+ 19d,
+ 23d,
+ 29d,
+ 31d,
+ 37d
};
database.createAndInitialize();
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 3d3f0276b..94f1ed45d 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
@@ -207,6 +207,156 @@ public class MySqlMetadataAccessorITCase extends MySqlSourceTestBase {
assertThat(actualSchema).isEqualTo(expectedSchema);
}
+ @Test
+ public void testMysql57PrecisionTypesSchema() {
+ fullTypesMySql57Database.createAndInitialize();
+
+ String[] tables = new String[] {"precision_types"};
+ MySqlMetadataAccessor metadataAccessor =
+ getMetadataAccessor(tables, fullTypesMySql57Database);
+
+ Schema actualSchema =
+ metadataAccessor.getTableSchema(
+ TableId.tableId(
+ fullTypesMySql57Database.getDatabaseName(), "precision_types"));
+ Schema expectedSchema =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .fromRowDataType(
+ RowType.of(
+ new DataType[] {
+ DataTypes.DECIMAL(20, 0).notNull(),
+ DataTypes.DECIMAL(6, 2),
+ DataTypes.DECIMAL(9, 4),
+ DataTypes.DECIMAL(20, 4),
+ DataTypes.TIME(0),
+ DataTypes.TIME(3),
+ DataTypes.TIME(6),
+ DataTypes.TIMESTAMP(0),
+ DataTypes.TIMESTAMP(3),
+ DataTypes.TIMESTAMP(6),
+ DataTypes.TIMESTAMP_LTZ(0),
+ DataTypes.TIMESTAMP_LTZ(3),
+ DataTypes.TIMESTAMP_LTZ(6),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE()
+ },
+ new String[] {
+ "id",
+ "decimal_c0",
+ "decimal_c1",
+ "decimal_c2",
+ "time_c",
+ "time_3_c",
+ "time_6_c",
+ "datetime_c",
+ "datetime3_c",
+ "datetime6_c",
+ "timestamp_c",
+ "timestamp3_c",
+ "timestamp6_c",
+ "float_c0",
+ "float_c1",
+ "float_c2",
+ "real_c0",
+ "real_c1",
+ "real_c2",
+ "double_c0",
+ "double_c1",
+ "double_c2",
+ "double_precision_c0",
+ "double_precision_c1",
+ "double_precision_c2"
+ }))
+ .build();
+ assertThat(actualSchema).isEqualTo(expectedSchema);
+ }
+
+ @Test
+ public void testMysql8PrecisionTypesSchema() {
+ fullTypesMySql8Database.createAndInitialize();
+
+ String[] tables = new String[] {"precision_types"};
+ MySqlMetadataAccessor metadataAccessor =
+ getMetadataAccessor(tables, fullTypesMySql8Database);
+
+ Schema actualSchema =
+ metadataAccessor.getTableSchema(
+ TableId.tableId(
+ fullTypesMySql8Database.getDatabaseName(), "precision_types"));
+ Schema expectedSchema =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .fromRowDataType(
+ RowType.of(
+ new DataType[] {
+ DataTypes.DECIMAL(20, 0).notNull(),
+ DataTypes.DECIMAL(6, 2),
+ DataTypes.DECIMAL(9, 4),
+ DataTypes.DECIMAL(20, 4),
+ DataTypes.TIME(0),
+ DataTypes.TIME(3),
+ DataTypes.TIME(6),
+ DataTypes.TIMESTAMP(0),
+ DataTypes.TIMESTAMP(3),
+ DataTypes.TIMESTAMP(6),
+ DataTypes.TIMESTAMP_LTZ(0),
+ DataTypes.TIMESTAMP_LTZ(3),
+ DataTypes.TIMESTAMP_LTZ(6),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE(),
+ DataTypes.DOUBLE()
+ },
+ new String[] {
+ "id",
+ "decimal_c0",
+ "decimal_c1",
+ "decimal_c2",
+ "time_c",
+ "time_3_c",
+ "time_6_c",
+ "datetime_c",
+ "datetime3_c",
+ "datetime6_c",
+ "timestamp_c",
+ "timestamp3_c",
+ "timestamp6_c",
+ "float_c0",
+ "float_c1",
+ "float_c2",
+ "real_c0",
+ "real_c1",
+ "real_c2",
+ "double_c0",
+ "double_c1",
+ "double_c2",
+ "double_precision_c0",
+ "double_precision_c1",
+ "double_precision_c2"
+ }))
+ .build();
+ assertThat(actualSchema).isEqualTo(expectedSchema);
+ }
+
private void testAccessDatabaseAndTable(UniqueDatabase database) {
database.createAndInitialize();
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 08d25a714..9699ed908 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
@@ -140,6 +140,18 @@ CREATE TABLE precision_types
timestamp_c TIMESTAMP(0) NULL,
timestamp3_c TIMESTAMP(3) NULL,
timestamp6_c TIMESTAMP(6) NULL,
+ float_c0 FLOAT(6, 0),
+ float_c1 FLOAT(20, 3),
+ float_c2 FLOAT(24, 12),
+ real_c0 REAL(6, 0),
+ real_c1 REAL(20, 3),
+ real_c2 REAL(24, 12),
+ double_c0 DOUBLE(6, 0),
+ double_c1 DOUBLE(20, 3),
+ double_c2 DOUBLE(24, 12),
+ double_precision_c0 DOUBLE PRECISION(6, 0),
+ double_precision_c1 DOUBLE PRECISION(20, 3),
+ double_precision_c2 DOUBLE PRECISION(24, 12),
PRIMARY KEY (id)
) DEFAULT CHARSET=utf8;
@@ -156,4 +168,16 @@ VALUES (DEFAULT,
'2020-07-17 18:00:22',
'2020-07-17 18:00',
'2020-07-17 18:00:22',
- '2020-07-17 18:00:22');
+ '2020-07-17 18:00:22',
+ 2,
+ 3,
+ 5,
+ 7,
+ 11,
+ 13,
+ 17,
+ 19,
+ 23,
+ 29,
+ 31,
+ 37);
\ 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 54c6c7170..8abe8868c 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
@@ -141,9 +141,21 @@ CREATE TABLE precision_types
datetime_c DATETIME(0),
datetime3_c DATETIME(3),
datetime6_c DATETIME(6),
- timestamp_c TIMESTAMP(0),
- timestamp3_c TIMESTAMP(3),
- timestamp6_c TIMESTAMP(6),
+ timestamp_c TIMESTAMP(0) NULL,
+ timestamp3_c TIMESTAMP(3) NULL,
+ timestamp6_c TIMESTAMP(6) NULL,
+ float_c0 FLOAT(6, 0),
+ float_c1 FLOAT(20, 3),
+ float_c2 FLOAT(24, 12),
+ real_c0 REAL(6, 0),
+ real_c1 REAL(20, 3),
+ real_c2 REAL(24, 12),
+ double_c0 DOUBLE(6, 0),
+ double_c1 DOUBLE(20, 3),
+ double_c2 DOUBLE(24, 12),
+ double_precision_c0 DOUBLE PRECISION(6, 0),
+ double_precision_c1 DOUBLE PRECISION(20, 3),
+ double_precision_c2 DOUBLE PRECISION(24, 12),
PRIMARY KEY (id)
) DEFAULT CHARSET=utf8;
@@ -160,4 +172,16 @@ VALUES (DEFAULT,
'2020-07-17 18:00:22',
'2020-07-17 18:00',
'2020-07-17 18:00:22',
- '2020-07-17 18:00:22');
+ '2020-07-17 18:00:22',
+ 2,
+ 3,
+ 5,
+ 7,
+ 11,
+ 13,
+ 17,
+ 19,
+ 23,
+ 29,
+ 31,
+ 37);