mxm commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4399500337
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -159,15 +159,10 @@ public void collect(DynamicRecord inputData) {
newData.f1,
isForward);
} else {
- // Shuffled records with immediateUpdate=false go to the update side
output
- int writerKey =
- hashKeyGenerator.generateKey(
- data,
- foundSchema.resolvedTableSchema() != null
- ? foundSchema.resolvedTableSchema()
- : data.schema(),
- foundSpec != null ? foundSpec : data.spec(),
- data.rowData());
+ // Shuffled records with immediateUpdate=false go to the update side
output.
+ // Key them with the record's own schema, spec and row: the raw row is
laid out by the
+ // record's schema, so the table's schema or spec cannot be used with
it.
+ int writerKey = hashKeyGenerator.generateKey(data);
Review Comment:
The change will always use schema and spec from the record and not from
table schemas / specs. The issue before was that it could mix record schema /
spec with the corresponding table schema / spec, which leads to undefined
behavior.
--
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]
