Can I use the output of multiple

.collect()

calls in a order dependency

start

argument? Like this:

@op(
  ins={"start": In(Nothing)}
)
def wait_for_many():
  pass

@gaph
def mostly_parallel():
   parallel_done = []
   parallel_done.append(has_dynamic_out.map(do_one_thing))
   parallel_done.append(has_dynamic_out.map(do_another_thing))
   parallel_done.append(has_dynamic_out.map(do_different_thing))
   wait_for_many(start=parallel_done)

When I try to do it as written above, I get:

DagsterInvalidDefinitionError: In @graph mostly_parallel, received a list containing an invalid type at index 0 for input "start" (passed by keyword) in op invocation wait_for_many. Lists can only contain the output from previous solid invocations or input mappings, received <class 'list'>

(actual op names changed to match my minimal example).

maybe try .extend() instead of .append()? it kind of looks like you might be getting a List[List[Output]] in parallel_done and trying to pass that to wait_for_many(). that being said I have no idea if that is possible or not, interested to hear if you get it working.

parallel_done.extend(has_dynamic_out.map(do_one_thing))

says

Attempted to iterate over an InvokedSolidDynamicOutputWrapper. This object represents the dynamic output "result" from the op 'do_one_thing'. Use the "map" method on this object to create downstream dependencies that will be cloned for each DynamicOut that is resolved at runtime.

And if I make it

.extend(...map().collect)

then I get the

received list

error again.

dang. could you restructure it a bit to have wait_for_many() take 3 inputs, and just pass the results of each collect() to wait_for_many as separate inputs?

Yeah, that seems to work OK.