A new approach to AutomationCondition.on_cron() using Freshness Policies

[stevenayers]

Hi all 👋 I'd like to get some feedback on an idea I've been cooking up, leveraging Freshness Policies for Declarative Automation. We use AutomationCondition.on_cron() as our de facto condition, but it can be quite inflexible without making significant customisations.

Problem Statement

on_cron seems like the go-to automation condition for most people. At first, it seems fairly straightforward:

• Materialise an asset once in a cron time window
• Wait for all upstreams to update before materialising

But the complexity comes in when:

• You have assets that execute at different times of the day, and have to start fiddling around with cron expressions to make sure upstreams materialise within the asset's time window, which really dampens the initial draw of "event-based scheduling means you don't need to worry about schedules lining up".
• You have upstreams which materialise on different frequencies, and need to start adding custom logic based on tags, metadata etc. to tell Dagster what an upstream's time window is.

Proposal

With the new Freshness Policies, and dg.AutomationCondition.freshness_passed(), dg.AutomationCondition.freshness_warned() and dg.AutomationCondition.freshness_failed() conditions, I think there's potential for a new condition, similar to on_cron but leveraging Freshness Policies instead of the cron time window for upstream assets.

Let's say we have a new freshness condition available in the public API:

import dagster as dg
from dagster._core.definitions.declarative_automation import FreshnessResultCondition
from dagster._core.definitions.freshness import FreshnessState

# Proposed AutomationCondition to add to the public API...
# but with a better name...
def on_cron_deps_fresh(cron_schedule: str, cron_timezone: str = "UTC") -> dg.AutomationCondition:
    deps_are_fresh = dg.AutomationCondition.all_deps_match(
        # Assets are considered fresh if:
        # - they don't have a freshness policy (NOT_APPLICABLE) ... allows for static assets and ref tables
        # - the freshness is PASS or WARN
        FreshnessResultCondition(state=FreshnessState.PASS)
        | FreshnessResultCondition(state=FreshnessState.WARN)
        | FreshnessResultCondition(state=FreshnessState.NOT_APPLICABLE)
        | dg.AutomationCondition.will_be_requested()
    )
    return (
        dg.AutomationCondition.in_latest_time_window()
        & dg.AutomationCondition.cron_tick_passed(cron_schedule, cron_timezone).since_last_handled()
        & deps_are_fresh
    ).with_label(f"on_cron_deps_fresh({cron_schedule}, {cron_timezone})")

With on_cron:

@dg.asset(
    automation_condition=dg.AutomationCondition.on_cron("@daily"),
    freshness_policy=dg.FreshnessPolicy.time_window(
        warn_window=timedelta(hours=24),
        fail_window=timedelta(hours=28)
    )
)
def daily_upstream(): ...

@dg.asset(
    automation_condition=dg.AutomationCondition.on_cron("@weekly"),
    freshness_policy=dg.FreshnessPolicy.time_window(
        warn_window=timedelta(days=7),
        fail_window=timedelta(days=7, hours=6)
    )
)
def weekly_upstream(): ...

my_asset_condition = (
    dg.AutomationCondition.on_cron("@daily").ignore(dg.AssetSelection.assets(["weekly_upstream"]))
    & dg.AutomationCondition.all_deps_updated_since_cron("@weekly").allow(dg.AssetSelection.assets(["weekly_upstream"]))
)

@dg.asset(
    automation_condition=my_asset_condition,
    freshness_policy=dg.FreshnessPolicy.cron(
        deadline_cron="@daily",
        lower_bound_delta=timedelta(hours=36),
    )
)
def my_asset(): ...

return dg.Definitions(assets=[daily_upstream, weekly_upstream, my_asset])

That's a simple example, but the on_cron customizations can get pretty confusing. We use an automation condition builder in our code, which builds something like:

my_asset_condition = (
    dg.AutomationCondition.on_cron("@daily").ignore(
        dg.AssetSelection.tag("freshness", "hourly") |
        dg.AssetSelection.tag("freshness", "daily") |
        dg.AssetSelection.tag("freshness", "weekly") |
        dg.AssetSelection.tag("freshness", "monthly") |
        dg.AssetSelection.tag("freshness", "static")
    ) &
    dg.AutomationCondition.all_deps_updated_since_cron("@hourly").allow(dg.AssetSelection.tag("freshness", "hourly")) &
    dg.AutomationCondition.all_deps_updated_since_cron("@daily").allow(dg.AssetSelection.tag("freshness", "daily")) &
    dg.AutomationCondition.all_deps_updated_since_cron("@weekly").allow(dg.AssetSelection.tag("freshness", "weekly")) &
    dg.AutomationCondition.all_deps_updated_since_cron("@monthly").allow(dg.AssetSelection.tag("freshness", "monthly"))
)

But with on_cron_deps_fresh, you could do:

@dg.asset(
    automation_condition=on_cron_deps_fresh("@daily"),
    freshness_policy=dg.FreshnessPolicy.time_window(
        warn_window=timedelta(hours=24),
        fail_window=timedelta(hours=28)
    )
)
def daily_upstream(): ...

@dg.asset(
    automation_condition=on_cron_deps_fresh("@weekly"),
    freshness_policy=dg.FreshnessPolicy.time_window(
        warn_window=timedelta(days=7),
        fail_window=timedelta(days=7, hours=6)
    )
)
def weekly_upstream(): ...

@dg.asset(
    automation_condition=on_cron_deps_fresh("@daily"),
    freshness_policy=dg.FreshnessPolicy.cron(
        deadline_cron="@daily",
        lower_bound_delta=timedelta(hours=36),
    )
)
def my_asset(): ...

return dg.Definitions(assets=[daily_upstream, weekly_upstream, my_asset])

Discussion

This would allow you to just define the expected cron schedule based on when you want that asset to run, define your freshness policies on your assets, and Dagster takes care of the rest! No more having to think about overlapping time windows, and instead an asset materializes on a cron schedule, as long as the dependencies are up to date 😄

I was going to raise a PR, but keen to see if anyone can help improve the design of the automation condition, or think of any pitfalls I've overlooked :)

[OwenKephart]

Hi @stevenayers ! Thanks for putting up this discussion / the associated PR. Overall, I'm certainly a fan of using freshness policies to determine scheduling behavior, but there are some vague questions in my mind that I want to think / talk through before landing on the exact form that should take:

• is freshness always sufficient for knowing if it's ok to kick off a downstream asset? my feeling is that sometimes you may have an asset that you want to update once a day (let's say at midnight), but you naturally expect it to take a few hours before it actually can kick off (it itself is waiting for upstreams). therefore you set up a freshness policy that says it's fresh until (e.g.) 3am. if we use this policy, an asset downstream of that one would potentially think it was ok to kick off exactly at midnight, because that upstream asset would still be considered fresh at that point in time. you could avoid this situation by setting up that upstream asset to transition to the warn state exactly at midnight, but this also feel like not the exact way most people would intuitively set up their policies.
• given the above issue, is there maybe some way to combine cron-based checks on the upstreams with freshness-based ones? to be honest I don't have an answer to this, just wanted to continue the discussion here

again, overall definitely makes sense to me that there should be some way of incorporating this information, just trying to figure out what the best way of doing that is

[stevenayers]

Thanks @OwenKephart, yeah, you are right. It's a bit tricky and something I've been trialling, but it seems to me that although using freshness as your dependency gives you more flexibility on upstream time windows, this approach introduces new problems.

In your example, we could include & ~dg.AutomationCondition.run_in_progress() in the dependency evaluation?

import dagster as dg
from dagster._core.definitions.declarative_automation.operands import FreshnessResultCondition
from dagster._core.definitions.freshness import FreshnessState

def on_cron_deps_fresh(cron_schedule: str, cron_timezone: str = "UTC") -> dg.AutomationCondition:
    not_stale = (
        FreshnessResultCondition(state=FreshnessState.PASS)
        | FreshnessResultCondition(state=FreshnessState.WARN)
    )
    no_freshness_policy = FreshnessResultCondition(state=FreshnessState.NOT_APPLICABLE)
    not_running = ~dg.AutomationCondition.run_in_progress()

    deps_are_fresh = dg.AutomationCondition.all_deps_match(
        (
            not_stale | no_freshness_policy | dg.AutomationCondition.will_be_requested()
        ) & not_running
    )
    return (
        dg.AutomationCondition.in_latest_time_window()
        & dg.AutomationCondition.cron_tick_passed(cron_schedule, cron_timezone).since_last_handled()
        & deps_are_fresh
    ).with_label(f"on_cron_deps_fresh({cron_schedule}, {cron_timezone})")

I feel like there's something else I'm missing here, another case that would be a problem, but I can't quite figure out what it is... do any other scenarios come to mind?