[Data] Added tracking of block serialization time

[alexeykudinkin]

Changes

1. Modified Ray Core's generator handling sequence to inject back object creation & serialization durations
2. Updated task_completion_time_excl_backpressure_s to track both UDF block generation time AND block serialization overhead

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request introduces tracking for block serialization time, which is a valuable addition for performance monitoring in Ray Data. The core change involves modifying the generator execution flow in _raylet.pyx to feed back serialization duration to the caller using gen.send(). This new metric is then integrated throughout the data stack, including BlockExecStats, OpRuntimeMetrics, and relevant operators like MapOperator and HashShuffleOperator. The tests have also been updated to cover these new metrics.

The implementation is clean and effective. The use of yield expressions to pass data back into generators is a good pattern for this use case. I have one minor suggestion to improve code clarity.

[gemini-code-assist]

The reset parameter in udf_time_s is not used in the function body; the timer self._udf_time_s is always reset to 0. Since the only call site in map_operator.py passes reset=True, the behavior is correct for now. However, to improve clarity and prevent potential misuse in the future, I suggest removing the reset parameter from the method signature and updating the call in map_operator.py:772 to map_transformer.udf_time_s().

Suggested change

	def udf_time_s(self, reset: bool) -> float:
	cur_time_s = self._udf_time_s
	self._udf_time_s = 0
	return cur_time_s
	def udf_time_s(self) -> float:
	cur_time_s = self._udf_time_s
	self._udf_time_s = 0
	return cur_time_s

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[edoakes]

From offline discussion, core change looks ok to me. Let's consider it a private API for now. I will follow up to add a disclaimer & context in code comments.

Defer to others to review the data changes. You may want to consider an alternative name to "serialization time" since it is not purely serialization, but also includes memory allocation & copy time. Perhaps "block_write_time_s" or "block_output_time_s"