Flink: Fix RANGE distribution ignoring user-specified equality fields#17276
Draft
thswlsqls wants to merge 1 commit into
Draft
Flink: Fix RANGE distribution ignoring user-specified equality fields#17276thswlsqls wants to merge 1 commit into
thswlsqls wants to merge 1 commit into
Conversation
The RANGE branch of HashKeyGenerator.getKeySelector() checked schema.identifierFieldIds() but passed the equalityFields parameter to equalityFieldKeySelector. Its NONE and HASH siblings both check equalityFields, so RANGE now does the same. equalityFields is resolved by DynamicSinkUtil.resolveEqualityFieldNames, which falls back to the schema identifier field names when the user sets none. The two checks therefore only disagreed when a user set equality fields on a schema without identifier fields: RANGE fell back to round-robin routing while DynamicWriter still wrote equality deletes for those fields. Generated-by: Claude Code
thswlsqls
marked this pull request as draft
July 17, 2026 11:16
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
1 participant
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.
Closes #17274
Summary
case RANGEinHashKeyGenerator.getKeySelector()testedschema.identifierFieldIds().isEmpty()but passed theequalityFieldsparameter toequalityFieldKeySelector. Siblingscase NONEandcase HASHtestequalityFields.isEmpty(); RANGE now matches.DynamicWriterstill emitted equality deletes for them; the"because there are no equality fields set"warning, left unchanged, now matches the condition.equalityFieldscomes fromDynamicSinkUtil.resolveEqualityFieldNames(record.equalityFields(), schema), which falls back toschema.identifierFieldNames()when unset, soequalityFields.isEmpty()impliesschema.identifierFieldIds().isEmpty().getKeySelectoris private with one caller, so only the user-specified case changes behavior.Testing done
TestHashKeyGenerator#rangeModeWithEqualityFields(RANGE, equality fields{"id"}, schema without identifier fields). Fails before the change, passes after.:iceberg-flink:iceberg-flink-2.1:test --tests "*TestHashKeyGenerator*"— 17 tests passed.*TestDynamicRecordProcessor*,*TestDynamicWriter*,*TestDynamicRecordWithConfig*,*TestTableUpdater*— 30 tests passed.spotlessCheck,checkstyleMain,checkstyleTestpassed. The full module:checkwas not run — its MiniCluster suites do not complete here. Flink is not a REVAPI module.