lukecwik commented on code in PR #17327:
URL: https://github.com/apache/beam/pull/17327#discussion_r872889905
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java:
##########
@@ -223,27 +323,26 @@ public void put(
throws Exception {
// Ignore timestamp for grouping purposes.
// The Pre-combine output will inherit the timestamp of one of its inputs.
- WindowedValue<Object> groupingKey =
- WindowedValue.of(
- keyCoder.structuralValue(value.getValue().getKey()),
- IGNORED,
- value.getWindows(),
- value.getPane());
-
- GroupingTableEntry entry =
- lruMap.compute(
- groupingKey,
- (key, tableEntry) -> {
- if (tableEntry == null) {
- tableEntry =
- new GroupingTableEntry(
- groupingKey, value.getValue().getKey(),
value.getValue().getValue());
- } else {
- tableEntry.add(value.getValue().getValue());
- }
- return tableEntry;
- });
- weight += entry.getWeight();
+ GroupingTableKey groupingKey =
+ new GroupingTableKey(
+ value.getValue().getKey(), value.getWindows(), value.getPane(),
keyCoder, keySizer);
+
+ lruMap.compute(
+ groupingKey,
+ (key, tableEntry) -> {
+ if (tableEntry == null) {
+ weight += groupingKey.getWeight();
Review Comment:
There are two cases.
* key == structural key, then:
* GroupingTableKey weight = key weight + windows weight + pane info weight
* GroupingTableEntry weight = reference weight + accumulator weight
* key != structural key, then:
* GroupingTableKey weight = structural key weight + windows weight + pane
info weight
* GroupingTableEntry weight = key weight + accumulator weight
--
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]
