[Feature][Jdbc] Add String type column split Support by charset-based splitting algorithm

Author: chenhongyu05

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcOptions.java

@@ -233,5 +233,8 @@ public interface JdbcOptions {
                     .withDescription("string_split_mode");
 
     Option<String> STRING_SPLIT_MODE_COLLATE =
-            Options.key("string_split_mode_collate").stringType().noDefaultValue().withDescription("string_split_mode_collate");
+            Options.key("string_split_mode_collate")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("string_split_mode_collate");
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSourceConfig.java

@@ -18,10 +18,10 @@
 package org.apache.seatunnel.connectors.seatunnel.jdbc.config;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitMode;
 
 import lombok.Builder;
 import lombok.Data;
-import org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitMode;
 
 import java.io.Serializable;
 import java.util.List;

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java

@@ -131,7 +131,8 @@ private List<ChunkRange> splitTableIntoChunks(
                 return evenlyColumnSplitChunks(table, splitColumnName, min, max, chunkSize);
             case STRING:
                 if (useCharsetBasedStringSplitter) {
-                    return charsetBasedColumnSplitChunks(table, splitColumnName, min, max, chunkSize);
+                    return charsetBasedColumnSplitChunks(
+                            table, splitColumnName, min, max, chunkSize);
                 } else {
                     return evenlyColumnSplitChunks(table, splitColumnName, min, max, chunkSize);
                 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/FixedChunkSplitter.java

@@ -192,7 +192,8 @@ private Collection<JdbcSourceSplit> getJdbcSourceSplits(
     @Override
     protected PreparedStatement createSplitStatement(JdbcSourceSplit split, TableSchema schema)
             throws SQLException {
-        if (SqlType.STRING.equals(split.getSplitKeyType().getSqlType()) && !useCharsetBasedStringSplitter) {
+        if (SqlType.STRING.equals(split.getSplitKeyType().getSqlType())
+                && !useCharsetBasedStringSplitter) {
             return createStringColumnSplitStatement(split);
         }
         if (split.getSplitStart() == null && split.getSplitEnd() == null) {

seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/CharsetBasedSplitterTest.java

@@ -17,10 +17,11 @@
 
 package org.apache.seatunnel.connectors.seatunnel.jdbc.source;
 
-import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import lombok.extern.slf4j.Slf4j;
+
 import java.math.BigInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -146,7 +147,12 @@ public void testPerformance() {
         long endTime = System.currentTimeMillis();
         long duration = endTime - startTime;
 
-        log.info("Executing " + iterations + " encoding/decoding operations took: " + duration + " milliseconds");
+        log.info(
+                "Executing "
+                        + iterations
+                        + " encoding/decoding operations took: "
+                        + duration
+                        + " milliseconds");
         log.info("Average time per operation: " + (double) duration / iterations + " milliseconds");
     }
 
@@ -180,4 +186,4 @@ public void testRandomStrings() {
             assertEquals(randomString, decoded.trim());
         }
     }
-}
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-7/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlSplitIT.java

@@ -563,7 +563,7 @@ public void testDynamicCharSplit() {
         configMap.put("password", MYSQL_PASSWORD);
         configMap.put("table_path", MYSQL_DATABASE + "." + MYSQL_TABLE);
         configMap.put("split.size", "10");
-        configMap.put("string_split_mode", "charsetBased");
+        configMap.put("string_split_mode", "charset_based");
 
         TablePath tablePathMySql = TablePath.of(MYSQL_DATABASE, MYSQL_TABLE);
         MySqlCatalog mySqlCatalog =
@@ -627,7 +627,7 @@ public void testFixedCharSplit() {
         configMap.put("user", MYSQL_USERNAME);
         configMap.put("password", MYSQL_PASSWORD);
         configMap.put("table_path", MYSQL_DATABASE + "." + MYSQL_TABLE);
-        configMap.put("string_split_mode", "charsetBased");
+        configMap.put("string_split_mode", "charset_based");
 
         TablePath tablePathMySql = TablePath.of(MYSQL_DATABASE, MYSQL_TABLE);
         MySqlCatalog mySqlCatalog =