I think a fan-in op from after a dynamic output mapping still runs even if the dynamic out is skipped. For example:

@op(
    out=DynamicOut(str),
)
def yield_each_test_id():
    for test_id in read_some_test_id_list_from_somewhere():
        yield DynamicOutput(value=test_id, mapping_key=test_id)

@op
def work(test_id: str):
    pass # where the real work would  be

@op
def wait_for_all(ops):
    pass

@graph
def my_graph():
    multiple_values = yield_each_test_id(test_id_list)
    many_outputs = multiple_values.map(some_intervening_op)
    all_done = wait_for_all(many_outputs.collect())

I'm finding that even if yield_each_test_id() returns nothing, and no work ops run, wait_for_all still runs. But I would expect wait_for_all to get skipped. Here's how it looks in Dagit: