jto opened a new pull request, #40609: URL: https://github.com/apache/beam/pull/40609
Addresses #40607. Two batch-only optimizations of how the Flink runner moves records between the network, Flink's sorter and its spill files. They are in one PR because the second builds on the first: `LazyKvWindowedValue` is created by the length-prefixed serializer. ## Copy length-prefixed batch records without decoding them Flink's batch sorter copies every record with `TypeSerializer#copy(DataInputView, DataOutputView)` when it spills to disk. Beam coders do not delimit their output, so `CoderTypeSerializer` implements this as a full decode and re-encode. In a profiled Scio batch job this accounted for about half of the task CPU. New batch-only `--lengthPrefixBatchRecords` option, **enabled by default** (disable with `--lengthPrefixBatchRecords=false`). `CoderTypeInformation` then creates a `LengthPrefixedCoderTypeSerializer` that writes each record's encoded length before it, so `copy()` becomes a byte copy. Streaming jobs and serializers for state and timers are unchanged, so savepoint compatibility is not affected. ## Decode the values of batch grouping inputs lazily Records sent to GroupByKey and Combine operators are decoded when they arrive from the network and immediately re-encoded into Flink's sorter, although only their key is needed on the way. In a profiled Scio batch job, this round trip took most of the receiving task thread's time. New batch-only `--lazyBatchGroupingInputs` option, **enabled by default** (disable with `--lazyBatchGroupingInputs=false`). The input of grouping operators is deserialized as a `LazyKvWindowedValue`, which keeps the whole encoded record and decodes the header (timestamp, windows, pane) and the value only when accessed. Untouched records are serialized again from their original bytes, and `KvToFlinkKeyKeySelector` reads the key from those bytes without decoding the value. Records Flink only moves around, e.g. when merging spill files, are read and written with a single copy each. ## Testing - New `LengthPrefixedCoderTypeSerializerTest`: round trips of back-to-back and nested records, `copy()` as a byte copy without invoking the coder, EOF handling, equality with and without the prefix, snapshots, and the serializer only being used in batch when enabled. - New `LazyKvWindowedValueTest`: windowing information and keys read without decoding the value, decoding the value once, equality with decoded values, re-serialization of untouched and mutated values, mixing lazy and eager records, unsupported coders, and lazy values only being used in batch for supported coders. 🤖 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]
