[core][experimental] Catch errors for DAG API

[stephanie-wang]

Why are these changes needed?

Catch unhandled errors in DAG API:

• when Ray actor tasks with multiple returns are used, throw error
• throw error when dag.execute inputs don't match the expected number of args or kwargs
• supporting binding kwargs to normal Python values
• throw error when user tries to bind DAG nodes to kwargs (not supported yet)
• When raising RayTaskError, raise it as instance of the original exception

Related issue number

Closes #46222.

[kevin85421]

Do you mean that the following example is invalid?

with InputNode() as dag_input:
    a = dag_input[0]
    ray_dag = add.bind(a, dag_input)

In my understanding, a is an InputAttributeNode. Is my understanding correct?

[stephanie-wang]

Yes, that's right!

The problem was that it's now unclear what dag_input should resolve to. Actually I'm not sure how the non-compiled DAG handles this, let me check.

[stephanie-wang]

Hmm seems that this is not handled well in non-compiled DAG either:

In [7]: @ray.remote
   ...: class Actor:
   ...:     def foo(self, *args):
   ...:         return args
   ...:     def bar(self):
   ...:         return self.foo()
   ...: 

In [8]: a = Actor.remote()

In [9]: with InputNode() as inp:
   ...:     dag = a.foo.bind(inp.foo, inp[0], inp)
   ...: 

In [10]: dag.execute(1, foo='hi')
Out[10]: ObjectRef(c2668a65bda616c10d375aa2d30b6f75e1f3eed60100000001000000)

In [11]: ray.get(dag.execute(1, foo="hi"))
Out[11]: ('hi', 1, <ray.dag.input_node.DAGInputData at 0x7f272264d340>)

[ruisearch42]

Put this to the end of _get_or_compile() since it is always called?

[stephanie-wang]

Prefer to do it separately because get_or_compile shouldn't need to depend on the inputs to dag.execute() (which can change each time dag.execute() is called).