Is there a way to define an OpDefinition that functions like a .collect for graphs with dynamic outs? I'm creating everything via config-based factory so can't use the .map/.collect from the documentation.

When I try to leverage MultiDependencyDefinition, I get this error: dagster.core.errors.DagsterInvalidDefinitionError: Currently, items in a fan-in dependency cannot be downstream of dynamic outputs. This seems strange since I thought the collect behavior is always downstream of dynamic outputs. Below is how the dependencies are generated via the configs.

dependencies = {}

for op_config in op_configs:
    def_name = op_config["op"]
    alias = op_config.get("alias", def_name)
    deps = op_config.get("deps", {}) or {}
    is_collect = op_config.get('is_collect')

    # set up op dependencies
    op_deps_entry = {}
    for input_name, output_data in deps.items():

        if is_collect:
            op_deps_entry[input_name] = MultiDependencyDefinition(
                [
                    DependencyDefinition(
                        output_data['op'],
                        output_data.get("output", "result")
                    ),
                ],
            )
        else:
            op_deps_entry[input_name] = DependencyDefinition(
                node=output_data["op"],
                output=output_data.get("output", "result"),
            )

    dependencies[
        NodeInvocation(name=def_name, alias=alias)
    ] = op_deps_entry

return GraphDefinition(
    name=graph_name,
    description=graph_description,
    node_defs=op_definitions,
    dependencies=dependencies,
)

Sorry I should have been more clear in my link -

DynamicCollectDependencyDefinition

is the thing you want. I just included

MultiDependencyDefinition

since it behaves in a similar way and is more well documented

works as expected now, thank you!