#3242 [BUG] Unpredictable inputs on conditional branch Issue created by frelyf Describe the bug Conditional branching doesn't pass input parameters consistently with deep nesting and high parallelization. Expected behavior Using the conditional branching method, if the check in .if_() has different variables than those used to call the method in .then() method, the parameters passed to the subworkflow should be the parameters in then(). But on occasion, the parameters passed are the ones in the .if_() method. This causes a type error of the parameter types are different, but could go unnoticed if the parameters are of the same type. In this case, n1.o0 is evaluated in .if_(). Chain, country, date, output_path are outputs from tasks and potentially passed to downstream tasks in each .then() method call. Expected input dict: {"chain":"dummy_chain","country":"dummy_country","date":"1/2/2022 120000 AM UTC","output_path":"gs://root_bucket/examples/helloflyte/greeting/v1/chain=dummy_chain/country=dummy_country/year=2022/month=01/day=02/"} Wrong output dict: {1 item n1.o0:false } Additional context to reproduce 1. Write a shelltask that takes input parameters 2. Write python tasks to generate the inputs for the shelltask (testableFlagTarget against GCStorage). 3. Determine whether the shelltask should be executed (e.g. input data is missing or output data already exists excludes execution and simply logs state). 3a. On the conditional branch, evaluate the data state with boolean comparison in .if_() 3b. If the condition is good, execute the shell task with the input. 4. Wrap the workflow in a launch plan and create a method that takes lists of input values to loop over the launch plan. 5. Run the workflow in GKE flyte environment. Behavior becomes increasingly aggravated as max parallelism is increased. Code sample with workflow and fanout methods

@workflow
def greeting_wf(kickoff_time: datetime, chain: str, country: str):
    date = kickoff_time_converter(kickoff_time=kickoff_time)
    output_check = generate_output_flag_check_greeting(
        date=kickoff_time, chain=chain, country=country
    )
    output_path = output_path_greeting(date=kickoff_time, chain=chain, country=country)

    conditional_flow = (
        conditional("flag_check")
        .if_(output_check.is_false())
        .then(
            greeting_shell_task(
                date=kickoff_time, chain=chain, country=country, output_path=output_path
            )
        )
        .elif_(output_check.is_true())
        .then(
            previous_completion(
            chain=chain, country=country, date=date, task_name="helloflyte_greeting"
            )
        )
        .else_()
        .fail("Something went wrong. Please contact DPAR.")
    )

    output_check >> conditional_flow

def generate_backfill_workflow(
    start_date: datetime, end_date: datetime, base_lp: LaunchPlan, **kwargs
) -> Workflow:
    if base_lp.schedule is None:
        raise ValueError("Backfill can only be created for scheduled launchplans")

    if isinstance(base_lp.schedule, CronSchedule):
        pass
    else:
        raise NotImplementedError("The launchplan schedule needs to be a cron schedule")

    if start_date >= end_date:
        raise ValueError("Start date should be greater than end date")

    sub_name = "_".join(kwargs.values())
    wf = Workflow(name=f"backfill-{base_lp.name}-{sub_name}")
    lp_iter = croniter(
        base_lp.schedule.cron_schedule.schedule,
        start_time=start_date,
        ret_type=datetime,
    )
    while True:
        next_start_date = lp_iter.get_next()
        if next_start_date > end_date:
            break
        wf.add_launch_plan(
            base_lp, kickoff_time=next_start_date, **kwargs
        ).with_overrides(
            node_name=f"{base_lp.name}_{sub_name}_{next_start_date.strftime('%Y-%m-%d')}"
        )

    return wf

def backfill_greeting(
    chain_list: List[str],
    country_list: List[str],
    start_date: datetime,
    num_days: int,
    **kwargs,
) -> Workflow:
    wf_wrapper = Workflow(name="helloflyte_greeting")
    end_date = start_date + timedelta(days=num_days)
    for chain in chain_list:
        for country in country_list:
            wf_wrapper.add_entity(
                generate_backfill_workflow(
                    start_date=start_date,
                    end_date=end_date,
                    base_lp=lp_greeting,
                    chain=chain,
                    country=country,
                    **kwargs,
                )
            ).with_overrides(name=f"greeting_{chain}_{country}")
    return wf_wrapper

Screenshots No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte