Flink: Avoid per-record Set allocation in the dynamic sink forward check - #17618
Open
sqd wants to merge 1 commit into
Open
Flink: Avoid per-record Set allocation in the dynamic sink forward check#17618sqd wants to merge 1 commit into
sqd wants to merge 1 commit into
Conversation
DynamicRecordProcessor.collect calls isForwardEligible for every record. That call resolves to DynamicSinkUtil.resolveEqualityFieldNames, which for a record with no user-supplied equality fields returns Schema.identifierFieldNames(), which uses Java Stream API to allocate and build a new HashSet. This HashSet is immediately discarded after isEmpty(). So each record on the hottest and commonest path (a table without identifier fields) allocates a set purely to ask whether it is empty. Ask the same question without materialising anything: test the user-supplied equality fields directly, then consult Schema.identifierFieldIds(), which Schema already memoizes as an ImmutableSet. Behaviour is unchanged and test still covers every branch.
Contributor
Author
|
Benchmarks: JMH, average time, JDK 21, 2 forks x 5 x 1 s, -prof gc, on an Apple M1 Max, against a 60-column schema:
For scale, an async-profiler CPU profile of a saturated TaskManager (AMD EPYC 9654P) running the dynamic sink in forward mode attributed 9.5% of the samples on the sink's task threads to this one check. |
anoopj
approved these changes
Aug 12, 2026
anoopj
left a comment
anoopj
left a comment
Member
There was a problem hiding this comment.
Code change looks right to me, and looks like there is coverage from existing tests.
Guosmilesmile
approved these changes
Aug 13, 2026
Guosmilesmile
left a comment
Guosmilesmile
left a comment
Contributor
There was a problem hiding this comment.
LGTM. This preserves the existing forward eligibility logic while avoiding the per-record set allocation. Thanks for the PR!
uros-b
approved these changes
Aug 13, 2026
Member
|
Thank you @sqd! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
DynamicRecordProcessor.collect calls isForwardEligible for every record. That call resolves to DynamicSinkUtil.resolveEqualityFieldNames, which for a record with no user-supplied equality fields returns Schema.identifierFieldNames(), which uses Java Stream API to allocate and build a new HashSet. This HashSet is immediately discarded after isEmpty(). So each record on the hottest and commonest path (a table without identifier fields) allocates a set purely to ask whether it is empty.
Ask the same question without materialising anything: test the user-supplied equality fields directly, then consult Schema.identifierFieldIds(), which Schema already memoizes as an ImmutableSet. Behaviour is unchanged and test still covers every branch.