@Scout how can I make it so that the asset only depends on partitions where the upstream asset has the same partition key? if it doesnt have the same partition key, it doesnt need to wait for it

To ensure that an asset only depends on partitions where the upstream asset has the same partition key, you can use the

IdentityPartitionMapping

. This partition mapping expects that the upstream and downstream assets are partitioned in the same way, and it maps partitions in the downstream asset to the same partition in the upstream asset. Here's an example from the documentation on how to set this up:

python from dagster import DailyPartitionsDefinition, AssetIn, asset, IdentityPartitionMapping  partitions_def = DailyPartitionsDefinition(start_date="2020-01-01")  @asset(partitions_def=partitions_def) def asset1():     ...  @asset(     partitions_def=partitions_def,     ins={         "asset1": AssetIn(             partition_mapping=IdentityPartitionMapping()         )     } ) def asset2(asset1):     ...

In this example, both

asset1

and

asset2

are defined with the same

DailyPartitionsDefinition

. The

IdentityPartitionMapping

is used to specify that

asset2

should only depend on the partition of

asset1

that has the same partition key. If a corresponding partition key does not exist in

asset1

,

asset2

will not wait for it.

@Scout How do I set that up with an automaterialization rule

In Dagster, auto-materialization rules help determine when an asset should be materialized based on various conditions. To set up an asset to only depend on partitions where the upstream asset has the same partition key, you can use the

AutoMaterializeRule.materialize_on_parent_updated

rule with a custom filter that ensures the partition keys match. Here's a general approach based on the documentation: 1. Define a custom filter function that checks if the upstream partition key matches the downstream partition key. 2. Use the

materialize_on_parent_updated

rule and pass the custom filter function to the

updated_parent_filter

parameter. The documentation does not provide a direct example of a custom filter function for this specific use case, but you can create a custom

AutoMaterializeAssetPartitionsFilter

that implements the desired logic. Here's a conceptual example of how you might set this up:

python from dagster import asset, AutoMaterializeRule, AutoMaterializeAssetPartitionsFilter  # Define your custom filter function def custom_partition_filter(upstream_partitions, downstream_partitions):     # Implement logic to filter upstream partitions based on downstream partition keys     # This function should return a list of partition keys that match the downstream partition keys     pass  # Create an instance of AutoMaterializeAssetPartitionsFilter using your custom filter function custom_filter = AutoMaterializeAssetPartitionsFilter(filter_fn=custom_partition_filter)  # Use the custom filter in your auto-materialization rule my_rule = AutoMaterializeRule.materialize_on_parent_updated(updated_parent_filter=custom_filter)  # Apply the rule to your asset @asset(auto_materialize=my_rule) def my_asset():     ...

Please note that the above code is a conceptual example and may not work as-is. The actual implementation of the

custom_partition_filter

function and how you apply the

AutoMaterializeRule

to your asset will depend on the specifics of your Dagster setup and the structure of your partitions. If you need more detailed guidance or if the current version of Dagster has introduced new features or best practices for achieving this, I recommend checking the latest documentation or reaching out to the Dagster community for assistance.

@Scout

AutoMaterializeAssetPartitionsFilter

does not have a

filter_fn

parameter

I apologize for the confusion. You are correct that

AutoMaterializeAssetPartitionsFilter

does not have a

filter_fn

parameter. The

AutoMaterializeAssetPartitionsFilter

is used to specify criteria for filtering partitions of upstream assets that trigger the materialization of downstream assets, but it does not allow for arbitrary filtering functions. Instead, you can use the

AutoMaterializeRule.skip_on_required_but_nonexistent_parents

rule to skip materializing an asset partition if it depends on parent partitions that do not exist. This rule is useful when you have a downstream asset that is time-partitioned and depends on an upstream asset that is also time-partitioned, but the upstream asset's partitions start at a later date. Here's an example of how you might use this rule:

python from dagster import asset, AutoMaterializeRule  # Apply the rule to your asset @asset(auto_materialize=AutoMaterializeRule.skip_on_required_but_nonexistent_parents()) def my_asset():     ...

This rule will ensure that the downstream asset's partitions are only materialized when the corresponding upstream partitions exist. If the upstream asset does not have a partition with the same key, the downstream asset will not wait for it and will skip materialization. If you need to enforce that an asset only depends on partitions with the same partition key and you cannot achieve this with the existing auto-materialization rules, you may need to implement custom logic within your asset definitions or use sensors to control the materialization process based on the existence of matching partition keys. For the most accurate and up-to-date information, please refer to the latest Dagster documentation or seek advice from the Dagster community.