jto opened a new pull request, #40611: URL: https://github.com/apache/beam/pull/40611
Addresses #40607. In batch, the final step of GroupByKey and Combine runs Beam's general windowing machinery (`WindowDoFnOperator` with `ReduceFnRunner`). For each key it binds keyed state, watermark holds and pane info, sets a timer, fires it when the key changes and clears all state. In a profiled Scio batch job this bookkeeping was 30-40% of the reduce phase, against about 6% for merging the accumulators. ## What changes New batch-only `--sortedBatchCombine` option, **enabled by default** (disable with `--sortedBatchCombine=false`). In the global window with the default trigger and without side inputs, the final step uses a new `SortedBatchCombineOperator`. It relies on Flink feeding a keyed operator all records of a key together (sorted batch execution), combines them in plain fields, and emits the result when the key changes or the input ends. Watermarks are held back until the end of the input, and the last key is emitted before the final watermark is forwarded. Other windowing strategies, triggers and side inputs keep using `WindowDoFnOperator`. ## Testing - New `SortedBatchCombineOperatorTest`: same output as `WindowDoFnOperator`, accumulator order, pipeline options and metrics available to the `CombineFn`, the last key emitted before the final watermark, and empty input. - `FlinkStreamingAggregationsTranslatorsTest`: the operator is only used in the supported cases. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
