[core] Update metadata in options properly

[suquark]

Why are these changes needed?

_metadata is a special keyword in Ray options that contains multiple namespaces. It is intended for config of different Ray libraries. However, current updating of options is incorrect for _metadata: it replaces all contents in _metadata, but what we really want is updating each namespace independently. For example:

old options: {"_metadata": {"workflow": {"a":1}}}
new options via ".options": {"_metadata": {"serve": {"b":2}}}
updated options: {"_metadata": {"serve": {"b":2}}}
expected result: {"_metadata": {"workflow": {"a":1}, "serve": {"b":2}}}

old options: {"_metadata": {"workflow": {"a":1}}}
new options via ".options": {"_metadata": {"workflow": {"b":2}}}
updated options: {"_metadata": {"workflow": {"b":2}}}
expected result: {"_metadata": {"workflow": {"a":1, "b":2}}}

This PR fixes this issue.

With this fix, we can support the follow syntax for updating workflow options:

@PublicAPI(stability="beta")
class options:
    """This class serves both as a decorator and options for workflow.

    Examples:
        >>> import ray
        >>> from ray import workflow
        >>>
        >>> # specify workflow options with a decorator
        >>> @workflow.options(catch_exceptions=True):
        >>> @ray.remote
        >>> def foo():
        >>>     return 1
        >>>
        >>> # speficy workflow options in ".options"
        >>> foo_new = foo.options(**workflow.options(catch_exceptions=False))
    """

    def __init__(self, **workflow_options: Dict[str, Any]):
        # TODO(suquark): More rigid arguments check like @ray.remote arguments. This is
        # fairly complex, but we should enable it later.
        valid_options = {
            "name",
            "metadata",
            "catch_exceptions",
            "max_retries",
            "allow_inplace",
            "checkpoint",
        }
        invalid_keywords = set(workflow_options.keys()) - valid_options
        if invalid_keywords:
            raise ValueError(
                f"Invalid option keywords {invalid_keywords} for workflow steps. "
                f"Valid ones are {valid_options}."
            )
        from ray.workflow.common import WORKFLOW_OPTIONS

        self.options = {"_metadata": {WORKFLOW_OPTIONS: workflow_options}}

    def keys(self):
        return ("_metadata",)

    def __getitem__(self, key):
        return self.options[key]

    def __call__(self, f: RemoteFunction) -> RemoteFunction:
        if not isinstance(f, RemoteFunction):
            raise ValueError("Only apply 'workflow.options' to Ray remote functions.")
        f._default_options.update(self.options)
        return f

Checks

• 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 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 :(

[fishbone]

I'm OK with this. I'll let @ericl make the final call.

[suquark]

The CI errors are about ValueError: Failed query HTTP endpoint. It seems not related to this PR, as previous tests for Ray options all pass, and all corresponding Linux tests passed. I'll merge it.