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

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

@@ -22,8 +22,8 @@
 import org.apache.seatunnel.api.sink.DataSaveMode;
 import org.apache.seatunnel.api.sink.SchemaSaveMode;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.dialectenum.FieldIdeEnum;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.source.StringSplitMode;
 
-import java.math.BigDecimal;
 import java.util.List;
 import java.util.Map;
 
@@ -171,14 +171,14 @@ public interface JdbcOptions {
                     .noDefaultValue()
                     .withDescription("partition column");
 
-    Option<BigDecimal> PARTITION_UPPER_BOUND =
+    Option<String> PARTITION_UPPER_BOUND =
             Options.key("partition_upper_bound")
-                    .bigDecimalType()
+                    .stringType()
                     .noDefaultValue()
                     .withDescription("partition upper bound");
-    Option<BigDecimal> PARTITION_LOWER_BOUND =
+    Option<String> PARTITION_LOWER_BOUND =
             Options.key("partition_lower_bound")
-                    .bigDecimalType()
+                    .stringType()
                     .noDefaultValue()
                     .withDescription("partition lower bound");
     Option<Integer> PARTITION_NUM =
@@ -225,4 +225,18 @@ public interface JdbcOptions {
                     .mapType()
                     .noDefaultValue()
                     .withDescription("additional connection configuration parameters");
+
+    Option<StringSplitMode> STRING_SPLIT_MODE =
+            Options.key("split.string_split_mode")
+                    .enumType(StringSplitMode.class)
+                    .defaultValue(StringSplitMode.SAMPLE)
+                    .withDescription(
+                            "Supports different string splitting algorithms. By default, `sample` is used to determine the split by sampling the string value. You can switch to `charset_based` to enable charset-based string splitting algorithm. When set to `charset_based`, the algorithm assumes characters of partition_column are within ASCII range 32-126, which covers most character-based splitting scenarios.");
+
+    Option<String> STRING_SPLIT_MODE_COLLATE =
+            Options.key("split.string_split_mode_collate")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Specifies the collation to use when string_split_mode is set to `charset_based` and the table has a special collation. If not specified, the database's default collation will be used.");
 }

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

@@ -18,6 +18,7 @@
 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;
@@ -44,6 +45,10 @@ public class JdbcSourceConfig implements Serializable {
     private int splitInverseSamplingRate;
     private boolean decimalTypeNarrowing;
 
+    private StringSplitMode stringSplitMode;
+
+    private String stringSplitModeCollate;
+
     public static JdbcSourceConfig of(ReadonlyConfig config) {
         JdbcSourceConfig.Builder builder = JdbcSourceConfig.builder();
         builder.jdbcConnectionConfig(JdbcConnectionConfig.of(config));
@@ -55,7 +60,8 @@ public static JdbcSourceConfig of(ReadonlyConfig config) {
                 config.getOptional(JdbcOptions.QUERY).isPresent()
                         && config.getOptional(JdbcOptions.PARTITION_COLUMN).isPresent();
         builder.useDynamicSplitter(!isOldVersion);
-
+        builder.stringSplitMode(config.get(JdbcOptions.STRING_SPLIT_MODE));
+        builder.stringSplitModeCollate(config.get(JdbcOptions.STRING_SPLIT_MODE_COLLATE));
         builder.splitSize(config.get(JdbcSourceOptions.SPLIT_SIZE));
         builder.splitEvenDistributionFactorUpperBound(
                 config.get(JdbcSourceOptions.SPLIT_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND));

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

@@ -27,7 +27,6 @@
 import lombok.experimental.Tolerate;
 
 import java.io.Serializable;
-import java.math.BigDecimal;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -53,10 +52,10 @@ public class JdbcSourceTableConfig implements Serializable {
     private Integer partitionNumber;
 
     @JsonProperty("partition_lower_bound")
-    private BigDecimal partitionStart;
+    private String partitionStart;
 
     @JsonProperty("partition_upper_bound")
-    private BigDecimal partitionEnd;
+    private String partitionEnd;
 
     @JsonProperty("use_select_count")
     private Boolean useSelectCount;

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

@@ -814,4 +814,44 @@ default boolean isSpecialDefaultValue(Object defaultValue, String sourceDialectN
     default String quotesDefaultValue(Object defaultValue) {
         return "'" + defaultValue + "'";
     }
+
+    default String getCollationSequence(Connection connection, String collate) {
+        StringBuilder sb = new StringBuilder();
+        String getDual = dualTable();
+        String baseQuery = "SELECT char_val FROM (";
+        StringBuilder unionQuery = new StringBuilder();
+        for (int i = 32; i <= 126; i++) {
+            if (i > 32) unionQuery.append(" UNION ALL ");
+            unionQuery.append("SELECT ? AS char_val ").append(getDual);
+        }
+        String sortedQuery =
+                baseQuery + unionQuery + ")  ndi_tmp_chars ORDER BY " + getCollateSql(collate);
+        log.info("sortedCollationQuery is " + sortedQuery);
+        PreparedStatement preparedStatement;
+        try {
+            preparedStatement = connection.prepareStatement(sortedQuery);
+            for (int i = 32; i <= 126; i++) {
+                log.debug("setString " + (i - 32) + " => " + (char) i);
+                preparedStatement.setString(i - 32 + 1, String.valueOf((char) i));
+            }
+
+            ResultSet resultSet = preparedStatement.executeQuery();
+            while (resultSet.next()) {
+                sb.append(resultSet.getString("char_val"));
+            }
+            return sb.toString();
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    default String getCollateSql(String collate) {
+        String getCollate =
+                StringUtils.isNotBlank(collate) ? "char_val COLLATE " + collate : "char_val";
+        return getCollate;
+    }
+
+    default String dualTable() {
+        return "";
+    }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/db2/DB2Dialect.java

@@ -98,4 +98,9 @@ public Optional<String> getUpsertStatement(
 
         return Optional.of(mergeStatement);
     }
+
+    @Override
+    public String dualTable() {
+        return " FROM SYSIBM.SYSDUMMY1 ";
+    }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleDialect.java

@@ -482,4 +482,24 @@ private boolean columnIsNullable(Connection connection, TablePath tablePath, Str
             return rs.getString("NULLABLE").equals("Y");
         }
     }
+
+    @Override
+    public String dualTable() {
+        return " FROM dual ";
+    }
+
+    @Override
+    public String getCollateSql(String collate) {
+        if (StringUtils.isNotBlank(collate)) {
+            StringBuilder sql = new StringBuilder();
+            sql.append("NLSSORT(")
+                    .append("char_val")
+                    .append(", 'NLS_SORT=")
+                    .append(collate)
+                    .append("')");
+            return sql.toString();
+        } else {
+            return "char_val";
+        }
+    }
 }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/vertica/VerticaDialect.java

@@ -22,6 +22,8 @@
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
@@ -107,4 +109,22 @@ public Optional<String> getUpsertStatement(
 
         return Optional.of(upsertSQL);
     }
+
+    /**
+     * <a
+     * href="https://docs.vertica.com/23.4.x/en/sql-reference/functions/data-type-specific-functions/string-functions/collation/">vertica-collation</a>
+     *
+     * @param collate
+     * @return
+     */
+    @Override
+    public String getCollateSql(String collate) {
+        if (StringUtils.isNotBlank(collate)) {
+            StringBuilder sql = new StringBuilder();
+            sql.append("COLLATION(").append("char_val").append(", '").append(collate).append("')");
+            return sql.toString();
+        } else {
+            return "char_val";
+        }
+    }
 }

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

@@ -85,6 +85,16 @@ public synchronized void close() {
         }
     }
 
+    protected static String filterOutUppercase(String str) {
+        StringBuilder sb = new StringBuilder();
+        for (char c : str.toCharArray()) {
+            if (!Character.isUpperCase(c)) {
+                sb.append(c);
+            }
+        }
+        return sb.toString();
+    }
+
     public Collection<JdbcSourceSplit> generateSplits(JdbcSourceTable table) throws Exception {
         log.info("Start splitting table {} into chunks...", table.getTablePath());
         long start = System.currentTimeMillis();