Upstream_output.definition_metadata doesn't contain the metadata since 1.7.11 for observable Source Assets [REGRESSION]

[ion-elgreco]

Dagster version

1.7.11

What's the issue?

Since v1.7.11 the InputContext.upstream_output.definition_metadata doesn't contain the metadata anymore of the input asset. This is quite problematic since we have IO managers that rely on this metadata of an asset.

with v1.7.11

with v1.7.10

What did you expect to happen?

No response

How to reproduce?

No response

Deployment type

None

Deployment details

No response

Additional information

No response

Message from the maintainers

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

[ion-elgreco]

@garethbrickman Can someone look into this? This breaks all IO managers implementations for source assets

[sverbruggen]

@garethbrickman Would really appreciate if if you could make some time for this, seems like a relatively major issue.

[OwenKephart]

Hi @sverbruggen @ion-elgreco , are either of you able to create a minimal reproduction of this issue? My attempts haven't been successful here, generally trying things along these lines:

def test_input_manager_with_source_assets() -> None:
    fancy_metadata = {"foo": "bar", "baz": 1.23}

    upstream = SourceAsset("upstream", metadata=fancy_metadata)

    @asset(ins={"upstream": AssetIn(input_manager_key="special_io_manager")})
    def downstream(upstream) -> int:
        return upstream + 1

    class MyIOManager(IOManager):
        def load_input(self, context) -> int:
            assert context.upstream_output is not None
            assert context.upstream_output.asset_key == AssetKey(["upstream"])
            assert context.upstream_output.definition_metadata == fancy_metadata

            return 2

        def handle_output(self, context, obj) -> None: ...

    defs = Definitions(
        assets=[upstream, downstream],
        resources={"special_io_manager": IOManagerDefinition.hardcoded_io_manager(MyIOManager())},
    )
    job = defs.get_implicit_job_def_for_assets([downstream.key])
    assert job is not None
    output = job.execute_in_process()

    assert output._get_output_for_handle("downstream", "result") == 3  # noqa: SLF001
    assert False

There was a change which resulted in those system-generated metadata keys (e.g. dagster/asset_execution_type) no longer appearing, but that wouldn't impact the user-provided metadata as you're experiencing.

[ion-elgreco]

@OwenKephart I have an MRE in my code base, will share it when I arrive back home tonight!

[OwenKephart]

Thank you!

[ion-elgreco]

@OwenKephart found a way to reproduce it with a slight modification of your example:

from dagster import (
    AssetIn,
    AssetKey,
    ConfigurableIOManager,  # noqa
    DataVersion,
    IOManager,  # noqa
    asset,
    materialize,
    observable_source_asset,
)


def test_input_manager_with_source_assets() -> None:
    fancy_metadata = {"foo": "bar", "baz": 1.23}

    @observable_source_asset(metadata=fancy_metadata)
    def upstream():
        return DataVersion('1')
    # upstream = SourceAsset("upstream", metadata=fancy_metadata)

    @asset(ins={"upstream": AssetIn(input_manager_key="special_io_manager")})
    def downstream(upstream) -> int:
        return upstream + 1

    class MyIOManager(IOManager):
        def load_input(self, context) -> int:
            assert context.upstream_output is not None
            assert context.upstream_output.asset_key == AssetKey(["upstream"])
            assert context.upstream_output.definition_metadata == fancy_metadata
            return 2

        def handle_output(self, context, obj) -> None: ...

    materialize(assets=[upstream,downstream], resources={"special_io_manager": MyIOManager()})

AssertionError: assert {'dagster/io_... 'io_manager'} == {'baz': 1.23, 'foo': 'bar'}
  Left contains 1 more item:
  {'dagster/io_manager_key': 'io_manager'}
  Right contains 2 more items:
  {'baz': 1.23, 'foo': 'bar'}
  Use -v to get more diff

The key is to use observable source assets, and materialize it instead of a job

[OwenKephart]

@ion-elgreco thanks for the reproduction, we should be able to get a fix in for next week's release

[ion-elgreco]

@OwenKephart thanks!! Much appreciated for the quick fix