fix: use record schema for partition key selector in HashKeyGenerator - #17685
Open
waterWang wants to merge 3 commits into
Open
fix: use record schema for partition key selector in HashKeyGenerator#17685waterWang wants to merge 3 commits into
waterWang wants to merge 3 commits into
Conversation
When the table schema is provided to generateKey(), the SelectorKey cache key was built using only the table schemaId, nulling the record's own schema. This caused the cache to return the same PartitionKeySelector for records with different writer schema variants, leading to ClassCastException when the accessor resolved a field ID from one variant against row data laid out per the other. Fix: always include the record's schema and spec in the SelectorKey, and use the record's own schema/spec when building the PartitionKeySelector, since the row data is organized per the record's schema, not the table schema. Fixes apache#17651
When the table schema is provided to generateKey(), the SelectorKey cache key was built using only the table schemaId, nulling the record's own schema. This caused the cache to return the same PartitionKeySelector for records with different writer schema variants, leading to ClassCastException when the accessor resolved a field ID from one variant against row data laid out per the other. Fix: always include the record's schema and spec in the SelectorKey, and use the record's own schema/spec when building the PartitionKeySelector, since the row data is organized per the record's schema, not the table schema. Fixes apache#17651
When the table schema is provided to generateKey(), the SelectorKey cache key was built using only the table schemaId, nulling the record's own schema. This caused the cache to return the same PartitionKeySelector for records with different writer schema variants, leading to ClassCastException when the accessor resolved a field ID from one variant against row data laid out per the other. Fix: always include the record's schema and spec in the SelectorKey, and use the record's own schema/spec when building the PartitionKeySelector, since the row data is organized per the record's schema, not the table schema. Fixes apache#17651
Contributor
|
Could you please provide unit tests? |
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
2 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.
Problem
When using Dynamic Sink with DistributionMode.HASH and records alternating between two writer schema variants (different field IDs), HashKeyGenerator's SelectorKey cache key uses only the table schema ID and nulls the record's schema. This causes the cache to return the same PartitionKeySelector for both variants, but the selector reads the wrong field position from the row data, producing ClassCastException.
Root Cause
Two issues in
HashKeyGenerator.java:SelectorKey cache key: When
tableSchemaIdis provided, the record's actualschemafield is nulled out (this.schema = tableSchemaId == null ? schema : null). Records with different dynamic schemas but the same table schema ID share the same cache key, returning a stale PartitionKeySelector.Key selector construction:
getKeySelector()is called witheffectiveSchema(the table schema) instead ofdynamicRecord.schema(). The row data is organized per the record's schema, so the key selector must be built with the record's schema.Fix
dynamicRecord.schema()anddynamicRecord.spec()when building the PartitionKeySelectorStack trace
Fixes #17651