damccorm commented on code in PR #41062:
URL: https://github.com/apache/beam/pull/41062#discussion_r4513404577
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -498,21 +501,24 @@ def add_input(self, accumulator, element, *args,
**kwargs):
def merge_accumulators(self, accumulators, *args, **kwargs):
result_heap = None
holds_comparables = None
+ use_comparables = self._compare is not operator.lt or bool(self._key)
+ n = self._n
for accumulator in accumulators:
holds_comparables, heap = accumulator
- if self._compare is not operator.lt or self._key:
+ if use_comparables:
heap = self._hydrated_heap(heap)
holds_comparables = True
else:
assert not holds_comparables
if result_heap is None:
- result_heap = heap
- else:
+ result_heap = list(heap)
+ elif n > 0:
for comparable in heap:
- _, result_heap = self.add_input(
- (holds_comparables, result_heap),
- comparable.value if holds_comparables else comparable)
Review Comment:
Note that this was getting rewrapped as a `cy_combiners.ComparableValue` in
`add_input`. With the new approach, we're avoiding this unwrapping/rewrapping
--
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]
