[Data] Use runtime object memory for scheduling

[bveeramani]

Why are these changes needed?

When selecting an operator to run, the scheduler doesn't consider how much object store memory an operator consumes. If an operator produces large blocks, the scheduler might select the operator too frequently, and your cluster can out of memory.

To prevent this from happening, this PR updates the scheduler so that it considers the incremental object store memory usage when selecting an operator.

Related issue number

Fixes #41190

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[bveeramani]

Will set this to 0 before merging. Currently enabled to ensure tests pass.

[raulchen]

Suggested change

	if (
	DataContext.get_current().use_runtime_metrics_scheduling
	and global_ok_sans_memory
	and op.metrics.average_bytes_change_per_task is not None
	and op.metrics.average_bytes_change_per_task <= 0
	):
	return True
	return global_ok_sans_memory and downstream_memory_ok
	if DataContext.get_current().use_runtime_metrics_scheduling:
	return (
	global_ok_sans_memory
	and op.metrics.average_bytes_change_per_task is not None
	and op.metrics.average_bytes_change_per_task <= 0
	)
	else:
	return global_ok_sans_memory and downstream_memory_ok

[bveeramani]

I tested this out with test_large_e2e_backpressure, and doing an if / else might not work. We're over the memory limit and average_bytes_change_per_task is initially None, so consume tasks aren't launched; as a result, we keep pulling data from the produce tasks until all of the produce tasks are complete.

Left this code as the original change for now. Hopefully that should minimize the risk, while still providing some benefit. In the long term, I think we need to devise a strategy for when metrics are None.