[FLINK-39749][mysql-cdc] Add scan.incremental.snapshot.string-key.compare-mode option

This commit introduces a new configuration option `scan.incremental.snapshot.string-key.compare-mode`
to fix chunk splitting and binlog event routing issues when MySQL collation differs from Java's
natural String ordering.

Problem:
- Java String.compareTo() is case-sensitive (Unicode code point order).
- MySQL collations like utf8mb4_general_ci are case-insensitive.
- This mismatch causes chunk boundaries computed by Java to diverge from actual MySQL row ordering,
  leading to premature unbounded chunks, overlapping splits, or lost binlog events.

Solution:
- Introduce ChunkKeyCompareMode enum: DEFAULT, CASE_INSENSITIVE, BINARY.
- DEFAULT: preserves existing behavior (String.compareTo()).
- CASE_INSENSITIVE: uses String.compareToIgnoreCase() for Java-side comparisons.
- BINARY: injects BINARY keyword in SQL predicates and uses byte-level comparison in Java.

Changes cover all three API layers:
- DataStream API (MySqlSourceBuilder)
- Flink SQL (MySqlTableSourceFactory)
- Pipeline YAML (MySqlDataSourceFactory)

Also updates documentation (EN + ZH) and adds test coverage.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: liziyan

docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md

@@ -441,6 +441,19 @@ Flink SQL> SELECT * FROM orders;
         如果跳过 backfill ，快照阶段捕获表的更改将在稍后的 binlog 读取阶段被回放，而不是合并到快照中。<br>
         警告：跳过 backfill 可能会导致数据不一致，因为快照阶段发生的某些 binlog 事件可能会被重放（仅保证 at-least-once ）。
         例如，更新快照阶段已更新的值，或删除快照阶段已删除的数据。这些重放的 binlog 事件应进行特殊处理。
+      </td>
+    </tr>
+    <tr>
+      <td>scan.incremental.snapshot.string-key.compare-mode</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">default</td>
+      <td>String</td>
+      <td>
+        增量快照阶段字符串类型 chunk key 的比较模式。可选值包括：
+        <li><code>default</code>：使用 Java 默认字符串比较（区分大小写，Unicode 码点顺序）。可能与 MySQL 的 <code>utf8mb4_general_ci</code> 等大小写不敏感排序规则不一致。</li>
+        <li><code>case-insensitive</code>：使用大小写不敏感比较（Java 中的 <code>compareToIgnoreCase</code>），适用于 MySQL 使用 <code>utf8mb4_general_ci</code> 等大小写不敏感排序规则的场景。</li>
+        <li><code>binary</code>：在 SQL 查询和 Java 比较中均使用二进制比较，强制按字节精确匹配。适用于 MySQL 使用 <code>utf8mb4_bin</code> 或 <code>BINARY</code> 关键字等二进制比较的场景。</li>
+      </td>
     </tr>
     <tr>
       <td>use.legacy.json.format</td>

docs/content.zh/docs/connectors/pipeline-connectors/mysql.md

@@ -340,6 +340,19 @@ pipeline:
         如果跳过 backfill ，快照阶段捕获表的更改将在稍后的 binlog 读取阶段被回放，而不是合并到快照中。<br>
         警告：跳过 backfill 可能会导致数据不一致，因为快照阶段发生的某些 binlog 事件可能会被重放（仅保证 at-least-once ）。
         例如，更新快照阶段已更新的值，或删除快照阶段已删除的数据。这些重放的 binlog 事件应进行特殊处理。
+      </td>
+    </tr>
+    <tr>
+      <td>scan.incremental.snapshot.string-key.compare-mode</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">default</td>
+      <td>String</td>
+      <td>
+        增量快照阶段字符串类型 chunk key 的比较模式。可选值包括：
+        <li><code>default</code>：使用 Java 默认字符串比较（区分大小写，Unicode 码点顺序）。可能与 MySQL 的 <code>utf8mb4_general_ci</code> 等大小写不敏感排序规则不一致。</li>
+        <li><code>case-insensitive</code>：使用大小写不敏感比较（Java 中的 <code>compareToIgnoreCase</code>），适用于 MySQL 使用 <code>utf8mb4_general_ci</code> 等大小写不敏感排序规则的场景。</li>
+        <li><code>binary</code>：在 SQL 查询和 Java 比较中均使用二进制比较，强制按字节精确匹配。适用于 MySQL 使用 <code>utf8mb4_bin</code> 或 <code>BINARY</code> 关键字等二进制比较的场景。</li>
+      </td>
     </tr>
     <tr>
       <td>metadata.list</td>

docs/content/docs/connectors/flink-sources/mysql-cdc.md

@@ -468,6 +468,18 @@ Only valid for cdc 1.x version. During a snapshot operation, the connector will
         For example updating an already updated value in snapshot, or deleting an already deleted entry in snapshot. These replayed change log events should be handled specially.
       </td>
     </tr>
+    <tr>
+      <td>scan.incremental.snapshot.string-key.compare-mode</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">default</td>
+      <td>String</td>
+      <td>
+        The compare mode for string type chunk key during incremental snapshot phase. Available values are:
+        <li><code>default</code>: Use Java's default string comparison (case-sensitive, Unicode code point order). This may be inconsistent with MySQL's case-insensitive collations like <code>utf8mb4_general_ci</code>.</li>
+        <li><code>case-insensitive</code>: Use case-insensitive comparison (<code>compareToIgnoreCase</code> in Java), suitable when MySQL uses case-insensitive collation such as <code>utf8mb4_general_ci</code>.</li>
+        <li><code>binary</code>: Use binary comparison in both SQL queries and Java comparison, forcing byte-level exact match. Suitable when MySQL uses binary collation like <code>utf8mb4_bin</code> or <code>BINARY</code> keyword.</li>
+      </td>
+    </tr>
     </tbody>
 </table>
 </div>

docs/content/docs/connectors/pipeline-connectors/mysql.md

@@ -362,6 +362,18 @@ pipeline:
         For example updating an already updated value in snapshot, or deleting an already deleted entry in snapshot. These replayed change log events should be handled specially.
       </td>
     </tr>
+    <tr>
+      <td>scan.incremental.snapshot.string-key.compare-mode</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">default</td>
+      <td>String</td>
+      <td>
+        The compare mode for string type chunk key during incremental snapshot phase. Available values are:
+        <li><code>default</code>: Use Java's default string comparison (case-sensitive, Unicode code point order). This may be inconsistent with MySQL's case-insensitive collations like <code>utf8mb4_general_ci</code>.</li>
+        <li><code>case-insensitive</code>: Use case-insensitive comparison (<code>compareToIgnoreCase</code> in Java), suitable when MySQL uses case-insensitive collation such as <code>utf8mb4_general_ci</code>.</li>
+        <li><code>binary</code>: Use binary comparison in both SQL queries and Java comparison, forcing byte-level exact match. Suitable when MySQL uses binary collation like <code>utf8mb4_bin</code> or <code>BINARY</code> keyword.</li>
+      </td>
+    </tr>
     <tr>
       <td>metadata.list</td>
       <td>optional</td>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java

@@ -30,6 +30,7 @@
 import org.apache.flink.cdc.common.source.DataSource;
 import org.apache.flink.cdc.common.utils.StringUtils;
 import org.apache.flink.cdc.connectors.mysql.source.MySqlDataSource;
+import org.apache.flink.cdc.connectors.mysql.source.config.ChunkKeyCompareMode;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
 import org.apache.flink.cdc.connectors.mysql.source.config.ServerIdRange;
 import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
@@ -80,6 +81,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_STRING_KEY_COMPARE_MODE;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
@@ -167,6 +169,9 @@ public DataSource createDataSource(Context context) {
         boolean useLegacyJsonFormat = config.get(USE_LEGACY_JSON_FORMAT);
         boolean isAssignUnboundedChunkFirst =
                 config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
+        ChunkKeyCompareMode chunkKeyCompareMode =
+                ChunkKeyCompareMode.fromValue(
+                        config.get(SCAN_INCREMENTAL_SNAPSHOT_STRING_KEY_COMPARE_MODE));
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -220,6 +225,7 @@ public DataSource createDataSource(Context context) {
                         .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
                         .useLegacyJsonFormat(useLegacyJsonFormat)
                         .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst)
+                        .chunkKeyCompareMode(chunkKeyCompareMode)
                         .skipSnapshotBackfill(skipSnapshotBackfill);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
@@ -358,6 +364,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(PARSE_ONLINE_SCHEMA_CHANGES);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+        options.add(SCAN_INCREMENTAL_SNAPSHOT_STRING_KEY_COMPARE_MODE);
         return options;
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java

@@ -224,6 +224,19 @@ public class MySqlDataSourceOptions {
                                     + "By default, the chunk key is the first column of the primary key."
                                     + "eg. db1.user_table_[0-9]+:col1;db[1-2].[app|web]_order_\\.*:col2;");
 
+    @Experimental
+    public static final ConfigOption<String> SCAN_INCREMENTAL_SNAPSHOT_STRING_KEY_COMPARE_MODE =
+            ConfigOptions.key("scan.incremental.snapshot.string-key.compare-mode")
+                    .stringType()
+                    .defaultValue("default")
+                    .withDescription(
+                            "The compare mode for string chunk key during incremental snapshot. Supported values are: "
+                                    + "'default' (uses Java String.compareTo), "
+                                    + "'case-insensitive' (uses Java String.compareToIgnoreCase to align with MySQL case-insensitive collations), "
+                                    + "'binary' (forces binary comparison in both MySQL SQL and Java). "
+                                    + "The 'case-insensitive' mode is recommended for tables using utf8mb4_general_ci or similar case-insensitive collations. "
+                                    + "The 'binary' mode ensures consistent byte-level ordering but may impact query performance during chunk splitting.");
+
     @Experimental
     public static final ConfigOption<Boolean> SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED =
             ConfigOptions.key("scan.incremental.close-idle-reader.enabled")

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -20,6 +20,7 @@
 import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 import org.apache.flink.cdc.connectors.mysql.debezium.task.MySqlBinlogSplitReadTask;
 import org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext;
+import org.apache.flink.cdc.connectors.mysql.source.config.ChunkKeyCompareMode;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
 import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
 import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
@@ -93,6 +94,7 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecords, MySqlSpl
             new StoppableChangeEventSourceContext();
     private final boolean isParsingOnLineSchemaChanges;
     private final boolean isBackfillSkipped;
+    private final ChunkKeyCompareMode chunkKeyCompareMode;
     private final Map<String, List<SourceRecord>> pendingSchemaChangeEvents;
 
     private static final long READER_CLOSE_TIMEOUT = 30L;
@@ -116,6 +118,7 @@ public BinlogSplitReader(StatefulTaskContext statefulTaskContext, int subtaskId)
         this.isParsingOnLineSchemaChanges =
                 statefulTaskContext.getSourceConfig().isParseOnLineSchemaChanges();
         this.isBackfillSkipped = statefulTaskContext.getSourceConfig().isSkipSnapshotBackfill();
+        this.chunkKeyCompareMode = statefulTaskContext.getSourceConfig().getChunkKeyCompareMode();
         this.pendingSchemaChangeEvents = new HashMap<>();
     }
 
@@ -316,7 +319,7 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
 
                 FinishedSnapshotSplitInfo matchedSplit =
                         SplitKeyUtils.findSplitByKeyBinary(
-                                finishedSplitsInfo.get(tableId), chunkKey);
+                                finishedSplitsInfo.get(tableId), chunkKey, chunkKeyCompareMode);
 
                 return matchedSplit != null && position.isAfter(matchedSplit.getHighWatermark());
             }
@@ -381,7 +384,12 @@ private void configureFilter() {
             }
             // Sort splits by splitStart for binary search optimization
             // Binary search requires sorted data to work correctly
-            splitsInfoMap.values().forEach(SplitKeyUtils::sortFinishedSplitInfos);
+            splitsInfoMap
+                    .values()
+                    .forEach(
+                            splits ->
+                                    SplitKeyUtils.sortFinishedSplitInfos(
+                                            splits, chunkKeyCompareMode));
         }
         this.finishedSplitsInfo = splitsInfoMap;
         this.maxSplitHighWatermarkMap = tableIdBinlogPositionMap;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java

@@ -360,7 +360,8 @@ public Iterator<SourceRecords> pollWithBuffer() throws InterruptedException {
                             currentSnapshotSplit.getSplitKeyType(),
                             nameAdjuster,
                             currentSnapshotSplit.getSplitStart(),
-                            currentSnapshotSplit.getSplitEnd());
+                            currentSnapshotSplit.getSplitEnd(),
+                            statefulTaskContext.getSourceConfig().getChunkKeyCompareMode());
                 }
             }
         }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlSnapshotSplitReadTask.java

@@ -247,7 +247,8 @@ private void createDataEventsForTable(
                         snapshotSplit.getTableId(),
                         snapshotSplit.getSplitKeyType(),
                         snapshotSplit.getSplitStart() == null,
-                        snapshotSplit.getSplitEnd() == null);
+                        snapshotSplit.getSplitEnd() == null,
+                        sourceConfig.getChunkKeyCompareMode());
         LOG.info(
                 "For split '{}' of table {} using select statement: '{}'",
                 snapshotSplit.splitId(),

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java

@@ -18,6 +18,7 @@
 package org.apache.flink.cdc.connectors.mysql.source;
 
 import org.apache.flink.cdc.common.annotation.PublicEvolving;
+import org.apache.flink.cdc.connectors.mysql.source.config.ChunkKeyCompareMode;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
 import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
 import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
@@ -312,6 +313,12 @@ public MySqlSourceBuilder<T> assignUnboundedChunkFirst(boolean assignUnboundedCh
         return this;
     }
 
+    /** The compare mode for string chunk key during incremental snapshot. Defaults to 'default'. */
+    public MySqlSourceBuilder<T> chunkKeyCompareMode(ChunkKeyCompareMode chunkKeyCompareMode) {
+        this.configFactory.chunkKeyCompareMode(chunkKeyCompareMode);
+        return this;
+    }
+
     /**
      * Build the {@link MySqlSource}.
      *