[Hotfix][Connector-Jdbc] Write MySQL to support set collection data type (apache#9553)

Co-authored-by: chestnufang <[email protected]>

Author: 2 people

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MySqlTypeConverter.java

@@ -75,6 +75,7 @@ public class MySqlTypeConverter implements TypeConverter<BasicTypeDefine<MysqlTy
     static final String MYSQL_LONGTEXT = "LONGTEXT";
     static final String MYSQL_JSON = "JSON";
     static final String MYSQL_ENUM = "ENUM";
+    static final String MYSQL_SET = "SET";
 
     // ------------------------------time-------------------------
     static final String MYSQL_DATE = "DATE";
@@ -243,6 +244,7 @@ public Column convert(BasicTypeDefine typeDefine) {
                 builder.scale(decimalUnsignedType.getScale());
                 break;
             case MYSQL_ENUM:
+            case MYSQL_SET:
                 builder.dataType(BasicType.STRING_TYPE);
                 if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
                     builder.columnLength(100L);

seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MySqlTypeConverterTest.java

@@ -1082,4 +1082,20 @@ public void testReconvertDatetimeForV55() {
         Assertions.assertEquals(MySqlTypeConverter.MYSQL_DATETIME, typeDefine.getColumnType());
         Assertions.assertEquals(MySqlTypeConverter.MYSQL_DATETIME, typeDefine.getDataType());
     }
+
+    @Test
+    public void testConvertSet() {
+        BasicTypeDefine<Object> typeDefine =
+                BasicTypeDefine.builder()
+                        .name("test")
+                        .columnType("SET('reading','sports','music','travel')")
+                        .dataType("SET")
+                        .length(3L)
+                        .build();
+        Column column = MySqlTypeConverter.DEFAULT_INSTANCE.convert(typeDefine);
+        Assertions.assertEquals(typeDefine.getName(), column.getName());
+        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
+        Assertions.assertEquals(3, column.getColumnLength());
+        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
+    }
 }