[damccorm]

Description

Optimizes five core combiners in sdks/python/apache_beam/transforms/combiners.py:

1. CountCombineFn.add_inputs: Checks isinstance(elements, Sized) to return accumulator + len(elements) in $O(1)$ and falls back to sum(1 for _ in elements) instead of allocating list(elements) (79.42x faster on sized batches).
2. ToListCombineFn.merge_accumulators and ConcatListCombineFn.merge_accumulators: Replaces quadratic sum(accumulators, []) with linear list(itertools.chain.from_iterable(accumulators)), and uses accumulator.extend(element) in ConcatListCombineFn.add_input (20.41x faster merging 500 100-item lists).
3. SampleCombineFn.add_input: Inlines reservoir min-heap threshold check (r > inner_heap[0][0]) to avoid tuple allocation and method dispatch when an element is not sampled (3.45x faster).
4. TopCombineFn.merge_accumulators: Pushes directly into result_heap via heappush / heappushpop instead of re-wrapping elements through self.add_input (3.44x faster).

Testing

• Verified with pytest sdks/python/apache_beam/transforms/combiners_test.py.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR introduces well-targeted, idiomatic performance optimizations across five built-in combiners (CountCombineFn, TopCombineFn, SampleCombineFn, ToListCombineFn, and ConcatListCombineFn). The changes eliminate quadratic list concatenations, redundant materializations of iterables, and unnecessary object allocations/method dispatches in hot loops while preserving exact semantics.

🔍 General Feedback

• All five optimizations are clean, safe, and adhere to CombineFn accumulator mutability contracts (notably copying the first accumulator's inner heap in TopCombineFn.merge_accumulators and creating fresh lists via itertools.chain.from_iterable).
• One minor edge case in TopCombineFn when n <= 0 is worth noting for consistency between add_input and merge_accumulators, though it is non-blocking.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🟢 Nit: If n <= 0 (e.g. TopCombineFn(0)), add_input will still push up to 1 element into heap because len(heap) < self._n is False and else: heapq.heappushpop(heap, comparable) is executed on an empty list (which pushes the element and pops it, leaving heap empty—wait: heappushpop([], x) actually returns x and leaves the list empty!). However, if the first accumulator in merge_accumulators somehow had elements or if n <= 0 is guarded in add_input as well (like in SampleCombineFn.add_input), guarding if self._n <= 0: return accumulator in TopCombineFn.add_input would make TopCombineFn(0) consistent and avoid unnecessary ComparableValue allocations.