damccorm opened a new pull request, #41062: URL: https://github.com/apache/beam/pull/41062
### 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`. -- 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]
