How do I feed multiple op outputs into a single op.

[dagsir[bot]]

Hi Dagster Community, is it possible to combine multiple streams of DynamicOutput into one op? e.g. pseudocode:

def op1():
    for i in range(1, 5):
        yield DynamicOutput(i, mapping_key=str(i))


@op(out=DynamicOut(int))
def op2():
    for i in range(6, 10):
        yield DynamicOutput(i, mapping_key=str(i))


@op
def op3(x, y):
    print(x + y)


@job()
def job1():
    xs = op1()
    ys = op2()
    xs.map(lambda x: ys.map(lambda y: op3(x, y)))```
I am getting error message like:
```dagster._core.errors.DagsterInvalidDefinitionError: op 'op3' cannot be downstream of more than one dynamic output. It is downstream of both "op2:result" and "op1:result"```
Thanks

---

*The question was originally asked in [Dagster Slack](https://dagster.slack.com/archives/C01U954MEER/p1700619146868899?thread_ts=1700619146.868899&cid=C01U954MEER).*

[benpankow]

Hello - right now, this functionality is not possible unless you collect() the outputs of at least one of the dynamic ops.