How to wait for any deep dependency in progress?

[PascalStehling]

Hi :)

The state of the Assets:
There is an Multi Asset which creates two source assets. These two source assets are quite big and take a while to get be processed by the IoManager. As the IoManager works sequentially, one after the other Asset is materialized.

There is now a small chain of processing of each of these Assets. At the end they get merged together again. This merging is quite compute intensive.

The problem:
As the source assets finish one after the other dagsters auto-materialization starts a processing chain for the first source asset, including the merging step. Shortly after that it starts a second chain for the second source asset, also including the merging asset.

The Question:
is it possible to configure the AutomationCondition in a way, that the merging Asset is run only once, by checking not just the direct parent but checking the whole chain recursively? Or that instead of two Automaterilization runs being created, only one is created when the latest asset is finished?

Example Code:
Just put the code in a file an run dagster dev -f FILENAME.py. Remember to activate the AutomationConditionSensor.

1. Materialize only the first asset (first two, left+right)
2. On the first run it will work like I would Imagine it, as it cannot create merged when right_2 was never materialized. Once everything was materialized once, it will run merged twice if you only run the first asset.

from time import sleep

from dagster import (
    AssetOut,
    AutomationCondition,
    ConfigurableIOManager,
    Definitions,
    InputContext,
    OutputContext,
    asset,
    multi_asset,
)

automation = AutomationCondition.eager()


class SaveData(ConfigurableIOManager):
    # specifies an optional string list input, via config system

    def handle_output(self, context: OutputContext, obj):
        print(context.asset_key.to_string())
        # Small fast upload
        if "left" in context.asset_key.to_string():
            return

        # big asset, long upload
        sleep(60)
        return

    def load_input(self, context: InputContext):
        print(context.asset_key)
        return context.asset_key.parts[0]


@multi_asset(
    outs={
        "left": AssetOut(automation_condition=automation, io_manager_key="io_manager"),
        "right": AssetOut(automation_condition=automation, io_manager_key="io_manager")
    })
def source() -> tuple[str, str]:
    return "left:", "right:"


@asset(automation_condition=automation)
def left_1(left):
    return left + " step 1"


@asset(automation_condition=automation)
def right_1(right):
    return right + " step 1"


@asset(automation_condition=automation)
def left_2(left_1):
    return left_1 + " - step 2"


@asset(automation_condition=automation)
def right_2(right_1):
    return right_1 + " - step 2"


@asset(automation_condition=automation)
def merged(left_2, right_2):
    return left_2 + "\n" + right_2


defs = Definitions(
    assets=[source, left_1, left_2, right_1, right_2, merged],
    resources={"io_manager": SaveData()},
)