Failing to materialize a single output of a multi asset fails the whole execution

[vladidobro]

What's the issue?

Hello Dagster!

It seems that when a single output of a multi asset fails during HANDLE_OUTPUT, the whole multi asset is aborted.
It seems that there is no way to continue the execution of the multiasset and materialize the other outputs in case one of them fails.

The obvious api would be

@multi_asset(outs={'a': AssetOut(), 'b': AssetOut()})
def mymultiasset(context):
    try:
        yield Output(None, output_name='a')
    except Exception as e:
        context.log.info('except')
        context.log.info(repr(e))
    finally:
        context.log.info('finally')
    context.log.info('after')
    yield Output(None, output_name='b')

but this just logs "finally" and then stops execution.
It seems that the executor is not throwing any errors to the execution function, as the "except" block was not executed.
I have a suspicion that the "finally" block executed only because of the implicit del destructor that generators with "try" or "with" get automatically, not because of dagster orchestration.

In any case, when (and if) the execution gets to the second "yield Output", dagster ignores it completely.

With single-assets, the only usecase for this is resource cleanup, such as

@asset
def myasset(context):
    with resource_that_needs_cleanup:
        yield Output(None)

But here it seems to work well, and the exit function of the context manager is always executed, probably also because of the implicit destructor.

I don't see very much into the internals of dagster, but IMHO the correct behaviour would be to throw the error inside the execution function generator and let it deal with it.

I understand that this is maybe more of a feature request than a bug, but it deviates from what would be naively expected and is undocumented (AFAIK), so I labelled it as a bug. Feel free to relabel it.

What did you expect to happen?

I would expect that the IOManager exception (bare, or wrapped in some IOManagerHandleOutputError(exc)) will be thrown to the asset execution function.
The execution function generator will be resumed with f.throw(exc).
It could decide to handle the error in its own way, optionally continuing to yield other outputs, or simply perform cleanup.
The failed output would be marked as failed.

How to reproduce?

from dagster import *

class IO(ConfigurableIOManager):
    def handle_output(self, context, obj):
        raise ValueError

    def load_input(self, context):
        ...

@multi_asset(
    can_subset=True,
    outs={
        'a': AssetOut(io_manager_key='io', is_required=False),
        'b': AssetOut(),
    }
)
def mymultiasset(context):
    try:
        yield Output(None, output_name='a')
    except Exception as e:
        context.log.info('except')
        context.log.info(repr(e))
    finally:
        context.log.info('finally')
    context.log.info('after')
    yield Output(None, output_name='b')

@asset(io_manager_key='io')
def myasset(context):
    try:
        yield Output(None)
    finally:
        context.log.info('ok')


job = define_asset_job(
    'job',
    [myasset],
)
multijob = define_asset_job(
    'multijob',
    [mymultiasset],
)

defs = Definitions(
        assets=[myasset, mymultiasset],
        jobs=[job, multijob],
        resources={'io': IO()},
)

Dagster version

dagster, version 1.10.1

Deployment type

Local

Deployment details

MacOS arm64

Additional information

No response

Message from the maintainers

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