[jira] [Created] (FLINK-38568) Performance bottleneck in BinlogSplitReader with large number of snapshot splits

["yuanfenghu (Jira)" <ji...@apache.org>]

yuanfenghu created FLINK-38568:
----------------------------------

             Summary: Performance bottleneck in BinlogSplitReader with large number of snapshot splits
                 Key: FLINK-38568
                 URL: https://issues.apache.org/jira/browse/FLINK-38568
             Project: Flink
          Issue Type: Improvement
          Components: Flink CDC
    Affects Versions: cdc-3.5.0
            Reporter: yuanfenghu
             Fix For: cdc-3.6.0


h2. Background

When using MySQL CDC connector with large tables split into thousands of chunks (e.g., 10,000+), the BinlogSplitReader.shouldEmit() method causes severe performance degradation.Performance Impact (observed in production):
 * CPU usage: 25.12% spent in splitKeyRangeContains()

 * 38,403 comparisons per binlog record

 * Algorithm: O(n) linear search through all finished snapshot splits

Root Causes:
 * Linear search: For each binlog record, the code iterates through all finished splits to find which split contains the record
 * Unsorted list: The finished splits list is not sorted, preventing optimization

h2. Solution

1. Sort splits by boundary in BinlogSplitReader.configureFilter():
{code:java}
splitsInfoMap.values().forEach(RecordUtils::sortFinishedSplitInfos); {code}
2. Replace linear search with binary search in BinlogSplitReader.shouldEmit():
{code:java}
for (FinishedSnapshotSplitInfo splitInfo : finishedSplitsInfo.get(tableId)) {    if (RecordUtils.splitKeyRangeContains(...)) { return true; }}
 

FinishedSnapshotSplitInfo matchedSplit =     RecordUtils.findSplitByKeyBinary(finishedSplitsInfo.get(tableId), chunkKey);return matchedSplit != null && position.isAfter(matchedSplit.getHighWatermark());{code}
h2. Performance Improvement
||Metric||Before||After||Improvement||
|Algorithm|O(n)|O(log n)|-|
|Comparisons|38,403|~16|2,400x|
|CPU usage|25.12%|<0.01%|2,500x|
|Time|34ms|<0.015ms|2,200x|

 
{code:java}
//代码占位符
`---ts=2025-10-24 18:09:07.334;thread_name=Source Data Fetcher for Source: MySQL Source -> Parse -> Side Output -> Case-insensitive Convert (1/4)#0;id=111;is_daemon=false;priority=5;TCCL=org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader@19a11164
    `---[136.386382ms] org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader:shouldEmit()
        +---[0.00% 0.002533ms ] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:isDataChangeRecord() #248
        +---[0.00% 0.002138ms ] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:getTableId() #249
        +---[0.01% 0.010575ms ] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:getBinlogPosition() #253
        +---[0.04% 0.04841ms ] org.apache.flink.cdc.connectors.mysql.debezium.reader.BinlogSplitReader:hasEnterPureBinlogPhase() #254
        +---[0.00% 6.7E-4ms ] org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext:getDatabaseSchema() #262
        +---[0.00% 0.006651ms ] io.debezium.connector.mysql.MySqlDatabaseSchema:tableFor() #262
        +---[0.00% 7.69E-4ms ] org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext:getSourceConfig() #263
        +---[0.00% 5.72E-4ms ] org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig:getChunkKeyColumns() #263
        +---[0.00% 6.17E-4ms ] org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext:getSourceConfig() #264
        +---[0.00% 4.89E-4ms ] org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig:isTreatTinyInt1AsBoolean() #264
        +---[0.02% 0.02135ms ] org.apache.flink.cdc.connectors.mysql.source.utils.ChunkUtils:getChunkKeyColumnType() #261
        +---[0.00% 0.002676ms ] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:getStructContainsChunkKey() #266
        +---[0.00% 6.1E-4ms ] org.apache.flink.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext:getSchemaNameAdjuster() #269
        +---[0.00% 0.003123ms ] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:getSplitKey() #268
        +---[13.63% min=4.54E-4ms,max=0.007806ms,total=18.587447ms,count=38403] org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo:getSplitStart() #272
        +---[13.80% min=4.55E-4ms,max=0.095948ms,total=18.823901ms,count=38403] org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo:getSplitEnd() #272
        +---[25.12% min=7.74E-4ms,max=0.038913ms,total=34.265124ms,count=38403] org.apache.flink.cdc.connectors.mysql.source.utils.RecordUtils:splitKeyRangeContains() #271
        +---[0.00% 9.15E-4ms ] org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo:getHighWatermark() #273
        `---[0.03% 0.042434ms ] org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset:isAfter() #273 {code}
 

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)