You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2023/11/30 11:23:37 UTC
(incubator-paimon) branch master updated: [cdc] Support longtext data type mapping to bytes (#2281)
This is an automated email from the ASF dual-hosted git repository.
yuzelin 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 28583f52c [cdc] Support longtext data type mapping to bytes (#2281)
28583f52c is described below
commit 28583f52c319d0e1a2ff62b2fc87fd9aac335ee2
Author: monster <60...@users.noreply.github.com>
AuthorDate: Thu Nov 30 19:23:31 2023 +0800
[cdc] Support longtext data type mapping to bytes (#2281)
---
docs/content/cdc-ingestion/overview.md | 7 ++-
.../shortcodes/generated/mysql_sync_database.html | 1 +
.../shortcodes/generated/mysql_sync_table.html | 1 +
.../paimon/flink/action/cdc/TypeMapping.java | 4 +-
.../flink/action/cdc/mysql/MySqlTypeUtils.java | 6 +-
.../cdc/mysql/MySqlCdcTypeMappingITCase.java | 68 ++++++++++++++++++++++
.../resources/mysql/type_mapping_test_setup.sql | 15 +++++
7 files changed, 97 insertions(+), 5 deletions(-)
diff --git a/docs/content/cdc-ingestion/overview.md b/docs/content/cdc-ingestion/overview.md
index da4a8a472..d945d8f10 100644
--- a/docs/content/cdc-ingestion/overview.md
+++ b/docs/content/cdc-ingestion/overview.md
@@ -89,9 +89,10 @@ behaviors of `RENAME TABLE` and `DROP COLUMN` will be ignored, `RENAME COLUMN` w
2. You can use type mapping option `to-nullable` (Use `--type-mapping`) to ignore all NOT NULL constraints (except primary keys).
3. You can use type mapping option `to-string` (Use `--type-mapping`) to map all MySQL data type to STRING.
4. You can use type mapping option `char-to-string` (Use `--type-mapping`) to map MySQL CHAR(length)/VARCHAR(length) types to STRING.
-5. MySQL BIT(1) type will be mapped to Boolean.
-6. When using Hive catalog, MySQL TIME type will be mapped to STRING.
-7. MySQL BINARY will be mapped to Paimon VARBINARY. This is because the binary value is passed as bytes in binlog, so it
+5. You can use type mapping option `longtext-to-bytes` (Use `--type-mapping`) to map MySQL LONGTEXT types to BYTES.
+6. MySQL BIT(1) type will be mapped to Boolean.
+7. When using Hive catalog, MySQL TIME type will be mapped to STRING.
+8. MySQL BINARY will be mapped to Paimon VARBINARY. This is because the binary value is passed as bytes in binlog, so it
should be mapped to byte type (BYTES or VARBINARY). We choose VARBINARY because it can retain the length information.
## Setting Custom Job Name
diff --git a/docs/layouts/shortcodes/generated/mysql_sync_database.html b/docs/layouts/shortcodes/generated/mysql_sync_database.html
index 0fc65d96a..6a24c8620 100644
--- a/docs/layouts/shortcodes/generated/mysql_sync_database.html
+++ b/docs/layouts/shortcodes/generated/mysql_sync_database.html
@@ -76,6 +76,7 @@ under the License.
</li>
<li>"to-string": maps all MySQL types to STRING.</li>
<li>"char-to-string": maps MySQL CHAR(length)/VARCHAR(length) types to STRING.</li>
+ <li>"longtext-to-bytes": maps MySQL LONGTEXT types to BYTES.</li>
</ul>
</td>
</tr>
diff --git a/docs/layouts/shortcodes/generated/mysql_sync_table.html b/docs/layouts/shortcodes/generated/mysql_sync_table.html
index 373818e14..55d2d26f4 100644
--- a/docs/layouts/shortcodes/generated/mysql_sync_table.html
+++ b/docs/layouts/shortcodes/generated/mysql_sync_table.html
@@ -56,6 +56,7 @@ under the License.
</li>
<li>"to-string": maps all MySQL types to STRING.</li>
<li>"char-to-string": maps MySQL CHAR(length)/VARCHAR(length) types to STRING.</li>
+ <li>"longtext-to-bytes": maps MySQL LONGTEXT types to BYTES.</li>
</ul>
</td>
</tr>
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java
index b6e22e74f..b2262baef 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/TypeMapping.java
@@ -64,13 +64,15 @@ public class TypeMapping implements Serializable {
* <li>TO_NULLABLE: ignores all NOT NULL constraints (except for primary keys).
* <li>TO_STRING: maps all MySQL types to STRING.
* <li>CHAR_TO_STRING: maps MySQL CHAR(length)/VARCHAR(length) types to STRING.
+ * <li>LONGTEXT_TO_BYTES: maps MySQL LONGTEXT types to BYTES.
* </ul>
*/
public enum TypeMappingMode {
TINYINT1_NOT_BOOL,
TO_NULLABLE,
TO_STRING,
- CHAR_TO_STRING;
+ CHAR_TO_STRING,
+ LONGTEXT_TO_BYTES;
private static final Map<String, TypeMappingMode> TYPE_MAPPING_OPTIONS =
Arrays.stream(TypeMappingMode.values())
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
index 4006fd38c..f87d08747 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlTypeUtils.java
@@ -46,6 +46,7 @@ import java.util.Map;
import java.util.Optional;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.CHAR_TO_STRING;
+import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.LONGTEXT_TO_BYTES;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TINYINT1_NOT_BOOL;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_STRING;
@@ -272,7 +273,6 @@ public class MySqlTypeUtils {
case TINYTEXT:
case TEXT:
case MEDIUMTEXT:
- case LONGTEXT:
case JSON:
case ENUM:
case GEOMETRY:
@@ -291,6 +291,10 @@ public class MySqlTypeUtils {
return length == null || length == 0
? DataTypes.VARBINARY(VarBinaryType.DEFAULT_LENGTH)
: DataTypes.VARBINARY(length);
+ case LONGTEXT:
+ return typeMapping.containsMode(LONGTEXT_TO_BYTES)
+ ? DataTypes.BYTES()
+ : DataTypes.STRING();
case TINYBLOB:
case BLOB:
case MEDIUMBLOB:
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
index 51e78afb2..5f03912d1 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
@@ -37,6 +37,7 @@ import java.util.stream.IntStream;
import static org.apache.paimon.flink.action.MultiTablesSinkMode.COMBINED;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.CHAR_TO_STRING;
+import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.LONGTEXT_TO_BYTES;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TINYINT1_NOT_BOOL;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_NULLABLE;
import static org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TO_STRING;
@@ -500,4 +501,71 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
Collections.singletonList("pk"));
}
}
+
+ @Test
+ @Timeout(60)
+ public void testLongtextToBytes() throws Exception {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name", "longtext_to_bytes_test");
+
+ MySqlSyncDatabaseAction action =
+ syncDatabaseActionBuilder(mySqlConfig)
+ .withMode(COMBINED.configString())
+ .withTypeMappingModes(LONGTEXT_TO_BYTES.configString())
+ .build();
+ runActionWithDefaultEnv(action);
+
+ FileStoreTable table = getFileStoreTable("t1");
+
+ try (Statement statement = getStatement()) {
+ statement.executeUpdate("USE longtext_to_bytes_test");
+
+ // test schema evolution
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), DataTypes.VARCHAR(10).notNull()
+ },
+ new String[] {"pk", "v1"});
+ waitForResult(
+ Collections.singletonList("+I[1, 1]"),
+ table,
+ rowType,
+ Collections.singletonList("pk"));
+
+ statement.executeUpdate("ALTER TABLE t1 ADD COLUMN v2 LONGTEXT");
+ statement.executeUpdate(
+ "INSERT INTO t1 VALUES (2, '2', 'This is an example of a long text string, meant to demonstrate the usage of the LONGTEXT data type in SQL databases.')");
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(10).notNull(),
+ DataTypes.BYTES()
+ },
+ new String[] {"pk", "v1", "v2"});
+ waitForResult(
+ Arrays.asList(
+ "+I[1, 1, NULL]",
+ "+I[2, 2, [84, 104, 105, 115, 32, 105, 115, 32, 97, 110, 32, 101, 120, 97, 109, 112, 108, 101, 32, 111, 102, 32, 97, 32, 108, 111, 110, 103, 32, 116, 101, 120, 116, 32, 115, 116, 114, 105, 110, 103, 44, 32, 109, 101, 97, 110, 116, 32, 116, 111, 32, 100, 101, 109, 111, 110, 115, 116, 114, 97, 116, 101, 32, 116, 104, 101, 32, 117, 115, 97, 103, 101, 32, 111, 102, 32, 116, 104, 101, 32, 76, 79, 78, 71, 84, 69, 88, 84, 32, 100, 97, 116, 97, 32, 116, 121, 112, 101, [...]
+ table,
+ rowType,
+ Collections.singletonList("pk"));
+
+ // test newly created table
+ statement.executeUpdate(
+ "CREATE TABLE _new_table (pk INT, v LONGTEXT, PRIMARY KEY (pk))");
+ statement.executeUpdate("INSERT INTO _new_table VALUES (1, 'Paimon')");
+
+ waitingTables("_new_table");
+ waitForResult(
+ Collections.singletonList("+I[1, [80, 97, 105, 109, 111, 110]]"),
+ getFileStoreTable("_new_table"),
+ RowType.of(
+ new DataType[] {DataTypes.INT().notNull(), DataTypes.BYTES()},
+ new String[] {"pk", "v"}),
+ Collections.singletonList("pk"));
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/type_mapping_test_setup.sql b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/type_mapping_test_setup.sql
index 471555442..9a65463aa 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/type_mapping_test_setup.sql
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/type_mapping_test_setup.sql
@@ -282,3 +282,18 @@ CREATE TABLE t1 (
);
INSERT INTO t1 VALUES (1, '1');
+
+-- ################################################################################
+-- testLongtextToBytes
+-- ################################################################################
+
+CREATE DATABASE longtext_to_bytes_test;
+USE longtext_to_bytes_test;
+
+CREATE TABLE t1 (
+ pk INT,
+ v1 VARCHAR(10) NOT NULL,
+ PRIMARY KEY (pk)
+);
+
+INSERT INTO t1 VALUES (1, '1');
\ No newline at end of file