asset_check is loading the wrong (non-executed (dynamic) partitioned asset)

[mtn331]

What's the issue?

I'm encountering an issue with asset checks when working with partitioned assets in Dagster. When materializing a specific partition (e.g., ShopCode=1000), the asset materializes correctly and saves to \artifacts\storage\transactions\1000.

However, when the asset check runs, it attempts to access different, non-materialized partitions (e.g., ShopCode=2000 or 3000) and fails before reaching the first context.log.info() statement.

The asset check should run against the specific partition that was materialized (ShopCode=1000) rather than attempting to access other partition files that haven't been materialized yet.

What did you expect to happen?

I'm wondering if this is the expected behavior for asset checks on partitioned assets, if they should automatically target the same partition that was just materialized, and if there's specific configuration needed to ensure asset checks run against the correct partition.

This is the error that occurs where I ran for partition key 1000.

dagster._core.errors.DagsterExecutionLoadInputError: Error occurred while loading input "transactions" of step "transactions_transactions_valid_shop": The above exception was caused by the following exception: FileNotFoundError: [Errno 2] No such file or directory: '..\\artifacts\\storage\\transactions\\2000'

How to reproduce?

This issue can be reproduced by running dagster dev, having an asset partitioned by "ShopCode", executing the asset for a specific partition, and observing that while the asset materializes correctly in the artifacts directory, the asset check runs against incorrect partition paths. The asset materialization works correctly, but asset checks are not targeting the correct partition. The csv file in this scenario is static and does not change.

from dagster import DynamicPartitionsDefinition, DagsterInstance
import polars as pl


partitions_shopcodes = DynamicPartitionsDefinition(name="shopcodes_partitions")

instance = DagsterInstance.get()
instance.add_dynamic_partitions(
    partitions_def_name='shopcodes_partitions',
    partition_keys =
    sorted((
        pl.scan_csv(
            r"csv_name",
            infer_schema_length=None,
            quote_char='"'
        )
        .select("ShopCode")
        .collect()
        .get_column("ShopCode")
        .unique()
        .to_list()
    ))
)

from dagster import asset, asset_check, AssetCheckExecutionContext, AssetCheckResult
@asset(        
    partitions_def=partitions_shopcodes
)
def transactions(context: AssetExecutionContext):

    df = pl.read_csv(some_path, infer_schema_length=None).filter(pl.col("ShopCode").eq(int(partition_key)))

    return Output(
        value=df
    )

@asset_check(asset=transactions, blocking=True)
def transactions_valid_shop(context: AssetCheckExecutionContext, transactions) -> AssetCheckResult:
    context.log.info("First log, this is not being reached")

    return AssetCheckResult(
        passed=True
    )

Dagster version

1.9.11

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.