dagster-dbt test failures not respected across asset definitions

Dagster version

1.5.12

What's the issue?

When defining different sets of @dbt_assets as part of a single job, dagster correctly identifies dependencies across them and correctly creates and orchestrated tasks within the job to execute downstream dependencies after the upstream assets have been materialised. In our case, we need to have different @dbt_assets definitions to correctly configure and run partitioned and non-partitioned assets as part of a single job.

However if one of the upstream assets' tests fail, it fails to recognise this when selecting downstream assets to execute, and will execute them independently of test success/failure.

If assets are defined within a single @dbt_assets definition, it correctly respects test failures and doesn't execute downstream assets.

What did you expect to happen?

Test failures are respected when deciding to execute downstream assets as part of a single job, whether the asset is defined in the same or across separate @dbt_assets definitions.

How to reproduce?

Create 4 dbt models where:

Upstream 1 - has a failing test, set a tag on it of "chain_1"
Downstream 1 - depends on Upstream 1, set a tag on it of "chain_1"
Upstream 2 - has a failing test, set a tag on it of "chain_1"
Dowstream 2 - has a failing test, set a tag on it of "chain_2"

Create 2 @dbt_asset definitions:

@dbt_assets(
    manifest=dbt_manifest_path,
    select="tag:chain_1",
)
def dbt_test_chain_one(
    context: AssetExecutionContext, dbt: DbtCliResource
):
    dbt_run_invocation = dbt.cli(["build"], context=context, raise_on_error=False)
    yield from dbt_run_invocation.stream()

@dbt_assets(
    manifest=dbt_manifest_path,
    select="tag:chain_2",
)
def dbt_test_chain_two(
    context: AssetExecutionContext, dbt: DbtCliResource
):
    dbt_run_invocation = dbt.cli(["build"], context=context, raise_on_error=False)
    yield from dbt_run_invocation.stream()

Create 1 job definition:

define_asset_job(name="test_deps", selection=[dbt_test_chain_one, dbt_test_chain_two])

Execute the job, and note that:

Upstream 1 + Upstream 2 will execute and fail their tests
Downstream 1 will not execute as the upstream test failed
Downstream 2 will execute despite the upstream test having failed

Deployment type

None

Deployment details

This occurs both in a local deployment + Dagster Helm Chart deployment

Additional information

You have 2 alternatives way to handle this as far as I'm aware:

You mark the upstream task as having failed (raise exception if model/test fails), and none of the downstreams in the second @dbt_asset definition will execute, even if their upstreams succeeded
You don't mark the task as failed, and all dowsntreams in the second @dbt_asset definition will execute

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.

dagster-io / dagster