Hi, I’m trying to define a non-partitioned graph asset that depends on the last partition of a partitioned upstream asset with something like this:

@op
def my_op(my_input):
   ...

@graph_asset(
    name="my_graph_asset",
    ins={
        "my_input": AssetIn(
            AssetKey(["my_upstream_asset"]),
            partition_mapping=LastPartitionMapping(),
        )
    },
)
def valuation_components(my_input):
    return my_op(my_input)

but it still tries to load all the partitions from the upstream in

my_op

. How to do this properly? (I guess I need to do something with my op, but I cannot figure out what.)

Which Dagster version are you on?

@yuhan I was trying this on 1.4.4