How to trigger downstream dbt assets for multiple upstream multi_assets with is_required=False?

[TyShkan]

Currently I have a standard dbt_assets definition and multiple upstream multi_assets with AssetOut(is_required=False).

If I do a full refresh of upstream assets, then dbt_assets trigger as expected.

In case if one or more assets of multi_asset don't yield their results, all of the downstream dbt assets would be skipped:
Skipping step dbt_assets due to skipped dependencies: ['one_of_upstream_multi_asset_name.asset_name']

What could be an approach to keep multi_assets with is_required=False flag and trigger downstream dbt assets as expected?

UPD: I've tried to #ask-ai, but no luck yet

[sryza]

@OwenKephart do you know if there's a way to do this? I think this may be a limitation of our current implementatin.

[OwenKephart]

@TyShkan Unfortunately this is a limitation of the current implementation -- the core execution logic will never start a downstream step if any of the upstream steps did not return a result.

One potential workaround would be to split your single dbt_assets method into multiple (ideally separated by which assets they're downstream of, although depending on the shape of your graph this may not be possible). This way, if one upstream asset is skipped, only the dbt assets downstream of it would actually be materialized.

[TyShkan]

@OwenKephart thanks! I’ve thought about it, in a form of creating assets for each dbt model with individual select, but it feels like a sloppy road. What could be a proper way to actually split one dbt_assets method into multiple?

[OwenKephart]

I think the best option down that road would (unfortunately) be a sort of factory approach, where you have some function in this general shape:

def get_dbt_assets(manifest) -> Sequence[AssetsDefinition]:

    ret = []
    for select in <some set of dbt selects that break up your graph into smaller pieces>:
        @dbt_assets(manifest=manifest, select=select, name=...)
        def _dbt_assets(...): ...
        
        ret.append(ret)
    return ret

[the4thamigo-uk]

I have also hit this issue. I think in my case I might have to force a materialization record for all upstream assets, even if nothing was changed.

@OwenKephart I would have thought that as long as the partition has been previously materialized for the upstream asset, then the dbt job should be able to go ahead?

[emirkmo]

Is this a limitation of the new @dbt_assets approach? I thought this was working fine with the old API?

Anyway, couldn't one create a job that materializes assets even if the upstream assets are partially materialized? Set this job to work on select downstream assets (or make it configurable) and create a sensor to trigger this job whenever one of the watched upstream assets fails to successfully materialize (might need to add some logic to wait until the upstream are fresh). Or somehow based on the freshness of the downstream asset.

[caelan-schneider]

Also dealing with this issue, but not for DBT assets. Basically I have two multi_assets, an upstream and a downstream, and the upstream has is_required=false for each AssetOut, while the downstream has internal_asset_deps mapped appropriately. But because the downstream still requires every asset from the upstream multi_asset to have been materialized, I can never materialize just a subset of assets.

I'm just going to use a factory pattern for now.

[the4thamigo-uk]

I also have hit an issue with skipping materialization, which may or may not be related.

#19994

[rexledesma]

We're tracking this feature request here now: #22242