This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0e4d4e82a [flink] add support for longtext , longblob for mysql cdc
(#944)
0e4d4e82a is described below
commit 0e4d4e82a6bf4d692917c1bc8444dc8af03d1238
Author: JunZhang <[email protected]>
AuthorDate: Thu Apr 20 17:37:50 2023 +0800
[flink] add support for longtext , longblob for mysql cdc (#944)
---
.../paimon/flink/action/cdc/mysql/MySqlTypeUtils.java | 2 ++
.../action/cdc/mysql/MySqlSyncTableActionITCase.java | 17 +++++++++++------
.../src/test/resources/mysql/setup.sql | 10 ++++++----
3 files changed, 19 insertions(+), 10 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
index 1c1f813db..7af5ee124 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
@@ -184,12 +184,14 @@ public class MySqlTypeUtils {
case VARCHAR:
return DataTypes.VARCHAR(Preconditions.checkNotNull(length));
case TEXT:
+ case LONGTEXT:
return DataTypes.STRING();
case BINARY:
return DataTypes.BINARY(Preconditions.checkNotNull(length));
case VARBINARY:
return DataTypes.VARBINARY(Preconditions.checkNotNull(length));
case BLOB:
+ case LONGBLOB:
return DataTypes.BYTES();
default:
throw new UnsupportedOperationException(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index b2f845177..cbc8772d5 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -424,9 +424,11 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
DataTypes.CHAR(10), // _char
DataTypes.VARCHAR(20), // _varchar
DataTypes.STRING(), // _text
+ DataTypes.STRING(), // _longtext
DataTypes.BINARY(10), // _bin
DataTypes.VARBINARY(20), // _varbin
- DataTypes.BYTES() // _blob
+ DataTypes.BYTES(), // _blob
+ DataTypes.BYTES() // _longblob
},
new String[] {
"_id",
@@ -479,9 +481,11 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
"_char",
"_varchar",
"_text",
+ "_longtext",
"_bin",
"_varbin",
- "_blob"
+ "_blob",
+ "_longblob"
});
FileStoreTable table = getFileStoreTable();
List<String> expected =
@@ -504,10 +508,11 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
+ "2023-03-23T14:30:05,
2023-03-23T14:30:05.123, 2023-03-23T14:30:05.123456, "
+ "2023-03-24T14:30, 2023-03-24T14:30:05.120, "
+ "2023-03-23T15:00:10.123456, "
- + "Paimon, Apache Paimon, Apache Paimon MySQL
Test Data, "
+ + "Paimon, Apache Paimon, Apache Paimon MySQL
Test Data, Apache Paimon MySQL Long Test Data, "
+ "[98, 121, 116, 101, 115, 0, 0, 0, 0, 0], "
+ "[109, 111, 114, 101, 32, 98, 121, 116, 101,
115], "
- + "[118, 101, 114, 121, 32, 108, 111, 110,
103, 32, 98, 121, 116, 101, 115, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97]"
+ + "[118, 101, 114, 121, 32, 108, 111, 110,
103, 32, 98, 121, 116, 101, 115, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97],
"
+ + "[108, 111, 110, 103, 32, 98, 108, 111, 98,
32, 98, 121, 116, 101, 115, 32, 116, 101, 115, 116, 32, 100, 97, 116, 97]"
+ "]",
"+I["
+ "2, 2.2, "
@@ -527,8 +532,8 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
+ "NULL, NULL, NULL, "
+ "NULL, NULL, "
+ "NULL, "
- + "NULL, NULL, NULL, "
- + "NULL, NULL, NULL"
+ + "NULL, NULL, NULL, NULL, "
+ + "NULL, NULL, NULL, NULL"
+ "]");
waitForResult(expected, table, rowType, Arrays.asList("pt", "_id"));
diff --git
a/paimon-flink/paimon-flink-common/src/test/resources/mysql/setup.sql
b/paimon-flink/paimon-flink-common/src/test/resources/mysql/setup.sql
index 715263831..6ff27fc15 100644
--- a/paimon-flink/paimon-flink-common/src/test/resources/mysql/setup.sql
+++ b/paimon-flink/paimon-flink-common/src/test/resources/mysql/setup.sql
@@ -117,10 +117,12 @@ CREATE TABLE all_types_table (
_char CHAR(10),
_varchar VARCHAR(20),
_text TEXT,
+ _longtext LONGTEXT,
-- BINARY
_bin BINARY(10),
_varbin VARBINARY(20),
_blob BLOB,
+ _longblob LONGBLOB,
PRIMARY KEY (_id)
);
@@ -159,9 +161,9 @@ INSERT INTO all_types_table VALUES (
-- TIMESTAMP
'2023-03-23 15:00:10.123456',
-- string
- 'Paimon', 'Apache Paimon', 'Apache Paimon MySQL Test Data',
+ 'Paimon', 'Apache Paimon', 'Apache Paimon MySQL Test Data', 'Apache Paimon
MySQL Long Test Data',
-- BINARY
- 'bytes', 'more bytes', 'very long bytes test data'
+ 'bytes', 'more bytes', 'very long bytes test data', 'long blob bytes test
data'
), (
2, 2.2,
NULL, NULL, NULL, NULL,
@@ -180,8 +182,8 @@ INSERT INTO all_types_table VALUES (
NULL, NULL, NULL,
NULL, NULL,
NULL,
- NULL, NULL, NULL,
- NULL, NULL, NULL
+ NULL, NULL, NULL, NULL,
+ NULL, NULL, NULL, NULL
);
CREATE TABLE incompatible_field_1 (
