[Data] Ray can't reconstruct inputs if Python garbage collects input references

[bveeramani]

What happened + What you expected to happen

I ran a batch inference job on spot instances. When GCP interrupted some instances, I expected Ray Data to recover, but instead my program errored with a message saying that the input objects are missing.

The issue might be caused by us removing references to input objects in the InputDataBuffer physical operator:

ray/python/ray/data/_internal/execution/operators/input_data_buffer.py

Lines 62 to 63 in b582905

	def _get_next_inner(self) -> RefBundle:
	return self._input_data.pop(0)

When we pop the object reference, the reference goes out scope, and Ray might garbage collect the object. So, when the object is later needed to reconstruct an output, Ray isn't able to find the input object.

Versions / Dependencies

bd9dc16

Reproduction script

Difficult to reproduce.

Issue Severity

High: It blocks me from completing my task.