You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "huyuanfeng2018 (via GitHub)" <gi...@apache.org> on 2025/10/27 06:45:30 UTC
[PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
huyuanfeng2018 opened a new pull request, #4166:
URL: https://github.com/apache/flink-cdc/pull/4166
## What is the purpose of the change
Optimize MySQL CDC binlog split lookup from O(n) to O(log n) using binary search, reducing CPU usage from 25% to nearly 0% for tables with thousands of splits.
## Brief change log
- Add `sortFinishedSplitInfos()` and `findSplitByKeyBinary()` methods
- Update `BinlogSplitReader` to use binary search instead of linear search
- Add comprehensive unit tests
## Performance Impact
- **Time Complexity**: O(n) → O(log n)
## Verifying this change
Added 6 new unit tests covering various scenarios including edge cases and consistency verification with linear search.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568] [mysql-cdc] [cdc-base] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3520313484
> CI failed. Apache Download CDN has already removed the Flink 1.20.1 binaries pkg. Perhaps we need to upgrade the dependency version.
@huyuanfeng2018 Could you append a commit to bump flink version to 1.20.3 to fix this issue ? I've thecked that https://dlcdn.apache.org/flink/flink-1.20.3/ should be okay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3521082273
@huyuanfeng2018 I like your community cooperation style, now we can rebase this PR to latest master and convert to normal one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3515204916
@huyuanfeng2018 please also add this improvement to base framework
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3450260138
hi , @ruanhang1993. Could you have the time to review this Pr? Thank you very much ~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on code in PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#discussion_r2517099692
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SourceRecordUtils.java:
##########
@@ -206,4 +175,165 @@ public static HistoryRecord getHistoryRecord(SourceRecord schemaRecord) throws I
String historyRecordStr = value.getString(HISTORY_RECORD_FIELD);
return new HistoryRecord(DOCUMENT_READER.read(historyRecordStr));
}
+
+ /**
+ * Sorts the given finished snapshot splits by their splitStart boundary in ascending order. The
+ * first split (splitStart == null) is treated as negative infinity, and the last split
+ * (splitEnd == null) is treated as positive infinity.
+ */
+ public static void sortFinishedSplitInfos(List<FinishedSnapshotSplitInfo> splits) {
+ if (splits == null || splits.size() <= 1) {
+ return;
+ }
+
+ splits.sort(
+ (leftSplit, rightSplit) -> {
+ Object[] leftSplitStart = leftSplit.getSplitStart();
+ Object[] rightSplitStart = rightSplit.getSplitStart();
+
+ if (leftSplitStart == null && rightSplitStart == null) {
+ return 0;
+ }
+ if (leftSplitStart == null) {
+ return -1;
+ }
+ if (rightSplitStart == null) {
+ return 1;
+ }
+
+ return compareSplit(leftSplitStart, rightSplitStart);
+ });
+ }
+
+ /**
+ * Uses binary search to find the split containing the specified key in a sorted split list.
+ *
+ * <p>IMPORTANT: The splits list MUST be sorted by splitStart before calling this method. Use
+ * sortFinishedSplitInfos() to sort the list if needed.
+ *
+ * <p>To leverage data locality for append-heavy workloads (e.g. auto-increment PKs), this
+ * method checks the first and last splits before applying binary search to the remaining
+ * subset.
+ *
+ * @param sortedSplits List of splits sorted by splitStart (MUST be sorted!)
+ * @param key The chunk key to search for
+ * @return The split containing the key, or null if not found
+ */
+ public static FinishedSnapshotSplitInfo findSplitByKeyBinary(
Review Comment:
Thanks for the quick review~
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3519861726
Thank you @loserwang1024 and @leonardBang for the review. I totally agree with this optimization for the auto-increment scenario. In this case, almost only one comparison is needed, which is very cool. At the same time, I have added this optimization in other sources as well, except for MongoDB and Oracle. Please take the time to review it again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3520322400
> > CI failed. Apache Download CDN has already removed the Flink 1.20.1 binaries pkg. Perhaps we need to upgrade the dependency version.
>
> @huyuanfeng2018 Could you append a commit to bump flink version to 1.20.3 to fix this issue ? I've thecked that https://dlcdn.apache.org/flink/flink-1.20.3/ should be okay.
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3520113199
CI failed. Apache Download CDN has already removed the Flink 1.20.1 binaries pkg. Perhaps we need to upgrade the dependency version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#discussion_r2517026124
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/utils/SourceRecordUtils.java:
##########
@@ -206,4 +175,165 @@ public static HistoryRecord getHistoryRecord(SourceRecord schemaRecord) throws I
String historyRecordStr = value.getString(HISTORY_RECORD_FIELD);
return new HistoryRecord(DOCUMENT_READER.read(historyRecordStr));
}
+
+ /**
+ * Sorts the given finished snapshot splits by their splitStart boundary in ascending order. The
+ * first split (splitStart == null) is treated as negative infinity, and the last split
+ * (splitEnd == null) is treated as positive infinity.
+ */
+ public static void sortFinishedSplitInfos(List<FinishedSnapshotSplitInfo> splits) {
+ if (splits == null || splits.size() <= 1) {
+ return;
+ }
+
+ splits.sort(
+ (leftSplit, rightSplit) -> {
+ Object[] leftSplitStart = leftSplit.getSplitStart();
+ Object[] rightSplitStart = rightSplit.getSplitStart();
+
+ if (leftSplitStart == null && rightSplitStart == null) {
+ return 0;
+ }
+ if (leftSplitStart == null) {
+ return -1;
+ }
+ if (rightSplitStart == null) {
+ return 1;
+ }
+
+ return compareSplit(leftSplitStart, rightSplitStart);
+ });
+ }
+
+ /**
+ * Uses binary search to find the split containing the specified key in a sorted split list.
+ *
+ * <p>IMPORTANT: The splits list MUST be sorted by splitStart before calling this method. Use
+ * sortFinishedSplitInfos() to sort the list if needed.
+ *
+ * <p>To leverage data locality for append-heavy workloads (e.g. auto-increment PKs), this
+ * method checks the first and last splits before applying binary search to the remaining
+ * subset.
+ *
+ * @param sortedSplits List of splits sorted by splitStart (MUST be sorted!)
+ * @param key The chunk key to search for
+ * @return The split containing the key, or null if not found
+ */
+ public static FinishedSnapshotSplitInfo findSplitByKeyBinary(
Review Comment:
Thanks @huyuanfeng2018 for the update, the updating looks good, one minor comment: this Utils class is a little long after this PR, could we extract a new Utils like SplitKeyUtils to make the code more readable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "huyuanfeng2018 (via GitHub)" <gi...@apache.org>.
huyuanfeng2018 commented on PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#issuecomment-3521116714
> @huyuanfeng2018 I like your community cooperation style, now we can rebase this PR to latest master and convert to normal one
Already rebased onto master. Waiting for CI to finish, I will change the PR status from Draft to Ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38568][mysql-cdc] Optimize binlog split lookup using binary search [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4166:
URL: https://github.com/apache/flink-cdc/pull/4166#discussion_r2513619463
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java:
##########
@@ -454,39 +453,153 @@ public static BinlogOffset getBinlogPosition(Map<String, ?> offset) {
/** Returns the specific key contains in the split key range or not. */
public static boolean splitKeyRangeContains(
Object[] key, Object[] splitKeyStart, Object[] splitKeyEnd) {
- // for all range
- if (splitKeyStart == null && splitKeyEnd == null) {
- return true;
+ return compareKeyWithRange(key, splitKeyStart, splitKeyEnd) == 0;
+ }
+
+ /**
+ * Sorts the list of FinishedSnapshotSplitInfo by splitStart in ascending order. This is
+ * required for binary search to work correctly.
+ *
+ * <p>Handles special cases: - Splits with null splitStart are considered as MIN value (sorted
+ * to front) - Splits with null splitEnd are considered as MAX value (sorted to back)
+ *
+ * <p>NOTE: Current implementation assumes single-field split keys (as indicated by
+ * getSplitKey()). If multi-field split keys are supported in the future, the comparison logic
+ * should be reviewed to ensure consistency with {@link
+ * #splitKeyRangeContains(Object[],Object[],Object[])}.
+ *
+ * @param splits List of splits to be sorted (sorted in-place)
+ */
+ public static void sortFinishedSplitInfos(List<FinishedSnapshotSplitInfo> splits) {
+ if (splits == null || splits.size() <= 1) {
+ return;
}
- // first split
- if (splitKeyStart == null) {
- int[] upperBoundRes = new int[key.length];
- for (int i = 0; i < key.length; i++) {
- upperBoundRes[i] = compareObjects(key[i], splitKeyEnd[i]);
+
+ splits.sort(
+ (leftSplit, rightSplit) -> {
+ Object[] leftSplitStart = leftSplit.getSplitStart();
+ Object[] rightSplitStart = rightSplit.getSplitStart();
+
+ // Splits with null splitStart should come first (they are the first split)
+ if (leftSplitStart == null && rightSplitStart == null) {
+ return 0;
+ }
+ if (leftSplitStart == null) {
+ return -1;
+ }
+ if (rightSplitStart == null) {
+ return 1;
+ }
+
+ // Compare split starts
+ return compareSplit(leftSplitStart, rightSplitStart);
+ });
+ }
+
+ /**
+ * Uses binary search to find the split containing the specified key in a sorted split list.
+ *
+ * <p>IMPORTANT: The splits list MUST be sorted by splitStart before calling this method. Use
+ * sortFinishedSplitInfos() to sort the list if needed.
+ *
+ * @param sortedSplits List of splits sorted by splitStart (MUST be sorted!)
+ * @param key The chunk key to search for
+ * @return The split containing the key, or null if not found
+ */
+ public static FinishedSnapshotSplitInfo findSplitByKeyBinary(
+ List<FinishedSnapshotSplitInfo> sortedSplits, Object[] key) {
+
+ if (sortedSplits == null || sortedSplits.isEmpty()) {
+ return null;
+ }
+
Review Comment:
Current PR is already in a good shape, I only have two possible suggestions:
(1) Considering that most tables use auto-increment primary keys and that INSERT operations outnumber UPDATE operations in the binlog, the majority of events in the binlog will typically correspond to either the last split or the first split. Leveraging this data locality, we can further optimize by first checking whether the changelog matches the first or last split and then performing binary search on the remaining splits.
(2) We can also do same optimization for IncrementalSourceStreamFetcher which is used by other cdc connectors
WDYT? @huyuanfeng2018
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org