Hello, does anyone have tried to make a conditional branching inside a "map" ? It seems like dagster does not take into account the

is_required=False

parameter

@solid(
    output_defs=[
        OutputDefinition(name="a", is_required=False),
        OutputDefinition(name="b", is_required=False),
        OutputDefinition(name="c", is_required=False)
    ]
)
def generate_branches(file_path: str):
    if 1 == 1:
        yield(None, output_name="a")
    else:
        yield(None, output_name="b")

@solid
def proc_file(file_path):
    branches = generate_branches(file_path=file_path)

@pipeline()
def my_pipeline():
    input_files = gather_files()

    output_files = input_files.map(proc_file)
    end(output_files.collect())

thanks for the report, seems like its just a bug to be fixed

I'll give it to you tonight. Are you able to reproduce? I can give you a working example

upon closer inspection - I believe the problem you are running in to is executing the

generate_branches

@solid

in the body of the

proc_file

@solid

So should I make

proc_file

a nested function inside the pipeline function ? How am I supposed to manage dependencies between solids? ( I've tried to use the

composite_solid

it does not solves the problem

So should I make

proc_file

a nested function inside the pipeline function ?

that or just remove the

@solid

decorator, but i suspect its more clear to make it an inline function

How am I supposed to manage deendencies between solids?

It looks like you are doing things generally right - so i might need a more specific example to understand where you are stuck

It seems to solve the problem. But can you use

collect

on

c.map(echo)

?

ya take a look at the second link https://dagster.phacility.com/D8533 which is the test i wrote trying to repro your report