Virtual Assets: Respect partitions definition of intermediate assets

[raayu83]

What's the use case?

It seems virtual assets don't respect the partitions definition of intermediate assets yet.
This would be useful to handled cases where the number of elements in a partition is reduced and you want to trigger using automation conditions.

Ideas of implementation

This is a minimal example that would work with the feature implemented.
When you trigger materialization of a single day for portal1 partition in upstream_ok asset, the downstream_asset would start automatically afterwards

import dagster as dg

# in the end we can't use virtual assets when we want to be able to reduce the partitions of an upstream asset
# the reason seems to be that has trouble resolving partition mappings through virtual intermediate assets
# even if the intermediate virtual asset is partitioned and has a multi partition mapping
# same goes if downstream asset has the multi partition mapping, since it might not have a portal partition

daily_partition = dg.DailyPartitionsDefinition(start_date="2026-01-01")


multi_partitions_detailed = dg.MultiPartitionsDefinition(
    {
        "date": daily_partition,
        "portal": dg.StaticPartitionsDefinition(["portal1", "portal2", "portal3"]),
    }
)

multi_partitions_reduced = dg.MultiPartitionsDefinition(
    {
        "date": daily_partition,
        "portal": dg.StaticPartitionsDefinition(["portal1"]),
    }
)


eager = (
    dg.AutomationCondition.eager()
    .without(dg.AutomationCondition.in_latest_time_window())
    .resolve_through_virtual()
)


@dg.asset(partitions_def=multi_partitions_detailed)
def upstream_ok(context: dg.AssetExecutionContext):
    context.log.info(context.partition_time_window)


@dg.asset(
    deps=[upstream_ok],
    is_virtual=True, # virtual assets don't work here currently
    automation_condition=eager,
    partitions_def=multi_partitions_reduced, 
)
def intermediate_ok(context: dg.AssetExecutionContext):
    pass


@dg.asset(
    partitions_def=daily_partition,
    deps=[intermediate_ok],
    automation_condition=eager,
)
def downstream_ok(context: dg.AssetExecutionContext):
    context.log.info(context.partition_time_window)

Additional information

No response

Message from the maintainers

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