erajio opened a new issue, #17651:
URL: https://github.com/apache/iceberg/issues/17651
### Apache Iceberg version
1.10.1
### Query engine
Flink
### Please describe the bug 🐞
We use the Dynamic Sink to ingest Confluent-Avro Kafka topics where multiple
writer schema versions are live simultaneously (normal schema-registry
evolution: some producers on version N, others on N+1). Per writer version we
derive an Iceberg Schema (independently fresh-assigned field IDs, so the same
field name has different IDs across variants) plus a PartitionSpec built
against that variant via
PartitionSpec.builderFor(variantSchema).identity("event_date"), and emit
DynamicRecords with DistributionMode.HASH.
When records for the same table alternate between two schema variants,
HashKeyGenerator's partition-key selector reads the wrong field position for
the partition source — the accessor resolves a field ID from one variant
against row data laid out per the other — producing:
Caused by: java.lang.ClassCastException: Cannot cast java.lang.String to
java.lang.Integer
at java.base/java.lang.Class.cast(Unknown Source)
at org.apache.iceberg.StructTransform.get(StructTransform.java:89)
at org.apache.iceberg.PartitionKey.get(PartitionKey.java:30)
at
org.apache.iceberg.PartitionSpec.partitionToPath(PartitionSpec.java:225)
at
org.apache.iceberg.flink.sink.PartitionKeySelector.getKey(PartitionKeySelector.java:64)
at
org.apache.iceberg.flink.sink.dynamic.HashKeyGenerator.lambda$partitionKeySelector$...(HashKeyGenerator.java:218)
at
org.apache.iceberg.flink.sink.dynamic.HashKeyGenerator.generateKey(HashKeyGenerator.java:104)
at
org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.emit(DynamicRecordProcessor.java:162)
(event_date is a DATE — internally Integer — and the accessor lands on a
String column from the other variant's layout.)
Reproduction conditions
- Table partitioned by identity on a date column that is not at the same
field ID/position across variants (fields differ before it in one variant).
- DynamicRecords alternate between variant A and variant B (each with its
own Schema + PartitionSpec instance, IDs assigned independently).
- DistributionMode.HASH. With DistributionMode.NONE the crash does not occur
(no partition-key extraction on the record path) — this is our current
workaround.
Happy to provide a full runnable reproducer if useful.
Expected behavior
Keying should be performed against the record's own schema/spec (or after
adaptation to the table schema), so mixed writer-schema versions — the normal
state of a schema-registry-fed stream — hash correctly.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
