Automaterialize all static partitions

[matthias-Q]

Hi,
we have a partitioned asset, which writes one file per partition (i.e. country). The policy is configured as:

    auto_materialize_policy=AutoMaterializePolicy.eager().with_rules(
        AutoMaterializeRule.materialize_on_cron(
            cron_schedule="0  4  *  *  1", all_partitions=True
        )

The upstream asset is refreshed every night. This triggers automatically the refresh of this partitioned asset, but only for one partition. What to I have to do, to automatically refresh all partitions.

[matthias-Q]

My partitioned Asset is as follow:

language_partitions = StaticPartitionsDefinition(
    [
        "GERMAN",
        "SPANISH",
        "FRENCH",
        "ENGLISH",
        "DUTCH",
        "HUNGARIAN",
        "ITALIAN",
        "POLISH",
    ]
)


@asset(
    partitions_def=language_partitions,
    deps={
        AssetKey(["upstream_asset1"]),
    },
    auto_materialize_policy=AutoMaterializePolicy.eager().with_rules(
        AutoMaterializeRule.materialize_on_cron(
            cron_schedule="0  4  *  *  1", all_partitions=True
        )
    ),
)
def partitioned_downstream_asset(context: AssetExecutionContext):
    context.log.info(f"Loading data for language {context.partition_key}")

    ...

When the upstream asset is materialized, only one downstream partitioned is triggered (for some random reason "SPANISH"). Even on the cron schedule, there is just one partition. Is this a bug? How do I have to configure it, to refresh all partitions?

[OwenKephart]

Hi @matthias-Q, to answer the question of why this wasn't working for you, there is a default rate limit on all AutoMaterializePolicies which prevents them from materializing more than one partition of an asset per tick. This is configurable, but it is recommended to use AutomationCondition in place of AutoMaterializePolicy: #22811

This would look like:

from dagster import (
    AssetExecutionContext,
    AssetKey,
    AutomationCondition,
    StaticPartitionsDefinition,
    asset,
)

language_partitions = StaticPartitionsDefinition(
    ["GERMAN", "SPANISH", "FRENCH", "ENGLISH", "DUTCH", "HUNGARIAN", "ITALIAN", "POLISH"]
)


@asset
def upstream_asset1() -> None: ...


@asset(
    partitions_def=language_partitions,
    deps={AssetKey(["upstream_asset1"])},
    automation_condition=AutomationCondition.on_cron("0 4 * * 1"),
)
def partitioned_downstream_asset(context: AssetExecutionContext):
    context.log.info(f"Loading data for language {context.partition_key}")

    ...