gustavodemorais commented on code in PR #27508:
URL: https://github.com/apache/flink/pull/27508#discussion_r2911972012
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java:
##########
@@ -64,13 +65,20 @@ public class AttributeBasedJoinKeyExtractor implements
JoinKeyExtractor, Seriali
// leftKeyExtractorsMap: extractors that read the left side (joined row so
far)
// using the same attribute order as in joinAttributeMap.
private final Map<Integer, List<KeyExtractor>> leftKeyExtractorsMap;
+ // RowData serializers for left key.
+ private final Map<Integer, RowDataSerializer> leftKeySerializersMap;
// rightKeyExtractorsMap: extractors to extract the right-side key from
each input.
private final Map<Integer, List<KeyExtractor>> rightKeyExtractorsMap;
+ // RowData serializers for right key.
+ private final Map<Integer, RowDataSerializer> rightKeySerializersMap;
// Data structures for the "common join key" shared by all inputs.
// Input 0 provides the canonical order and defines commonJoinKeyType.
private final Map<Integer, List<KeyExtractor>> commonJoinKeyExtractors;
+ // RowData serializers for common join key.
+ private final Map<Integer, RowDataSerializer> commonJoinKeySerializersMap;
private RowType commonJoinKeyType;
+ private boolean requiresKeyDeepCopy;
Review Comment:
Can you add a short comment to document why we added this functionality?
```suggestion
// Controls whether key rows built are serialized and copied. This is
required for the
// Heap State Backend to prevent object reuse issues, which can lead to data
corruption
private boolean requiresKeyDeepCopy;
```
--
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]
