You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/08/07 09:54:08 UTC
(flink-cdc) branch master updated: [FLINK-38164][pipeline-connector/mysql] support mysql long and long varchar type (#4076)
This is an automated email from the ASF dual-hosted git repository.
kunni 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 190915d4e [FLINK-38164][pipeline-connector/mysql] support mysql long and long varchar type (#4076)
190915d4e is described below
commit 190915d4e47c835d68827d65ff9565da37d1b1f6
Author: Chao Zhang <11...@users.noreply.github.com>
AuthorDate: Thu Aug 7 17:54:03 2025 +0800
[FLINK-38164][pipeline-connector/mysql] support mysql long and long varchar type (#4076)
---------
Co-authored-by: zhangchao.doovvv <zh...@bytedance.com>
---
docs/content.zh/docs/connectors/pipeline-connectors/mysql.md | 2 ++
docs/content/docs/connectors/pipeline-connectors/mysql.md | 2 ++
.../org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java | 4 ++++
.../flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java | 6 +++++-
.../cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java | 6 +++++-
.../src/test/resources/ddl/column_type_test.sql | 5 ++++-
.../src/test/resources/ddl/column_type_test_mysql8.sql | 5 ++++-
7 files changed, 26 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
index 1d6a338e6..fc38481fd 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
@@ -596,6 +596,8 @@ source:
TEXT<br>
MEDIUMTEXT<br>
LONGTEXT<br>
+ LONG<br>
+ LONG VARCHAR<br>
</td>
<td>STRING</td>
<td></td>
diff --git a/docs/content/docs/connectors/pipeline-connectors/mysql.md b/docs/content/docs/connectors/pipeline-connectors/mysql.md
index 3cd68675c..a8964cdaf 100644
--- a/docs/content/docs/connectors/pipeline-connectors/mysql.md
+++ b/docs/content/docs/connectors/pipeline-connectors/mysql.md
@@ -609,6 +609,8 @@ Notice:
TEXT<br>
MEDIUMTEXT<br>
LONGTEXT<br>
+ LONG<br>
+ LONG VARCHAR<br>
</td>
<td>STRING</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 05a23f7cc..0c59ec976 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
@@ -83,6 +83,8 @@ public class MySqlTypeUtils {
private static final String MEDIUMTEXT = "MEDIUMTEXT";
private static final String TEXT = "TEXT";
private static final String LONGTEXT = "LONGTEXT";
+ private static final String LONG_VARCHAR = "LONG VARCHAR";
+ private static final String LONG = "LONG";
private static final String DATE = "DATE";
private static final String TIME = "TIME";
private static final String DATETIME = "DATETIME";
@@ -217,6 +219,8 @@ public class MySqlTypeUtils {
case TEXT:
case MEDIUMTEXT:
case LONGTEXT:
+ case LONG_VARCHAR:
+ case LONG:
case JSON:
case ENUM:
case GEOMETRY:
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 c8546a3db..1e9191819 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
@@ -468,7 +468,9 @@ class MySqlFullTypesITCase extends MySqlSourceTestBase {
BinaryStringData.fromString(expectMultipointJsonText),
BinaryStringData.fromString(expectMultilineJsonText),
BinaryStringData.fromString(expectMultipolygonJsonText),
- BinaryStringData.fromString(expectGeometryCollectionJsonText)
+ BinaryStringData.fromString(expectGeometryCollectionJsonText),
+ BinaryStringData.fromString("long"),
+ BinaryStringData.fromString("long varchar")
};
// skip CreateTableEvent
@@ -679,6 +681,8 @@ class MySqlFullTypesITCase extends MySqlSourceTestBase {
DataTypes.STRING(),
DataTypes.STRING(),
DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
DataTypes.STRING());
private static final RowType JSON_TYPES =
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 33bb57d61..3fe1c664a 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
@@ -465,6 +465,8 @@ class MySqlMetadataAccessorITCase extends MySqlSourceTestBase {
DataTypes.STRING(),
DataTypes.STRING(),
DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
DataTypes.STRING()
},
new String[] {
@@ -520,7 +522,9 @@ class MySqlMetadataAccessorITCase extends MySqlSourceTestBase {
"multipoint_c",
"multiline_c",
"multipolygon_c",
- "geometrycollection_c"
+ "geometrycollection_c",
+ "long_c",
+ "long_varchar_c",
}))
.build();
assertThat(actualSchema).isEqualTo(expectedSchema);
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 e7dc07e28..53107ec59 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
@@ -72,6 +72,8 @@ CREATE TABLE common_types
multiline_c MULTILINESTRING,
multipolygon_c MULTIPOLYGON,
geometrycollection_c GEOMETRYCOLLECTION,
+ long_c LONG,
+ long_varchar_c LONG VARCHAR,
PRIMARY KEY (id)
) DEFAULT CHARSET=utf8;
@@ -94,7 +96,8 @@ VALUES (DEFAULT, 127, 255, 255, 32767, 65535, 65535, 8388607, 16777215, 16777215
ST_GeomFromText('MULTIPOINT((1 1),(2 2))'),
ST_GeomFromText('MultiLineString((1 1,2 2,3 3),(4 4,5 5))'),
ST_GeomFromText('MULTIPOLYGON(((0 0, 10 0, 10 10, 0 10, 0 0)), ((5 5, 7 5, 7 7, 5 7, 5 5)))'),
- ST_GeomFromText('GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))'));
+ ST_GeomFromText('GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))'),
+ 'long','long varchar');
CREATE TABLE time_types
(
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 92548381f..de0574109 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
@@ -72,6 +72,8 @@ CREATE TABLE common_types
multiline_c MULTILINESTRING,
multipolygon_c MULTIPOLYGON,
geometrycollection_c GEOMETRYCOLLECTION,
+ long_c LONG,
+ long_varchar_c LONG VARCHAR,
PRIMARY KEY (id)
) DEFAULT CHARSET=utf8;
@@ -94,7 +96,8 @@ VALUES (DEFAULT, 127, 255, 255, 32767, 65535, 65535, 8388607, 16777215, 16777215
ST_GeomFromText('MULTIPOINT((1 1),(2 2))'),
ST_GeomFromText('MultiLineString((1 1,2 2,3 3),(4 4,5 5))'),
ST_GeomFromText('MULTIPOLYGON(((0 0, 10 0, 10 10, 0 10, 0 0)), ((5 5, 7 5, 7 7, 5 7, 5 5)))'),
- ST_GeomFromText('GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))'));
+ ST_GeomFromText('GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))'),
+ 'long','long varchar');
CREATE TABLE time_types
(